Spaces:
Running
Running
| """GUS -> GSD context propagation.""" | |
| from __future__ import annotations | |
| from typing import Any, Dict, List | |
| CONTEXT_BUS_VERSION = "1.0" | |
| KNOWN_PROFILE_FIELDS = { | |
| "nip", | |
| "name", | |
| "pkd", | |
| "pkd_codes", | |
| "size", | |
| "voivodeship", | |
| "region", | |
| "address", | |
| "krs", | |
| "entity_type", | |
| } | |
| def build_context_payload(external_context: dict | None) -> dict: | |
| external_context = external_context or {} | |
| company_data = dict(external_context.get("company_data") or {}) | |
| gaps = list( | |
| external_context.get("data_gaps") | |
| or external_context.get("gaps") | |
| or [] | |
| ) | |
| return { | |
| "company_data": company_data, | |
| "gaps": gaps, | |
| "registry_enrichment": external_context.get("registry_enrichment") or {}, | |
| "credibility_flags": external_context.get("credibility_flags") or {}, | |
| "msp_analysis": external_context.get("msp_analysis") or {}, | |
| } | |
| def propagate_to_gsd_blackboard(blackboard: dict | None, external_context: dict | None) -> dict: | |
| bb = dict(blackboard or {}) | |
| payload = build_context_payload(external_context) | |
| company_data = payload.get("company_data") or {} | |
| if company_data: | |
| merged_company = dict(bb.get("company_data") or {}) | |
| for key, value in company_data.items(): | |
| if value not in (None, "", [], {}): | |
| merged_company[key] = value | |
| bb["company_data"] = merged_company | |
| gaps = payload.get("gaps") or [] | |
| if gaps: | |
| merged_gaps = list(bb.get("data_gaps") or []) | |
| for gap in gaps: | |
| if gap not in merged_gaps: | |
| merged_gaps.append(gap) | |
| bb["data_gaps"] = merged_gaps | |
| if payload.get("registry_enrichment"): | |
| bb["registry_enrichment"] = payload["registry_enrichment"] | |
| if payload.get("credibility_flags"): | |
| bb["credibility_flags"] = payload["credibility_flags"] | |
| if payload.get("msp_analysis"): | |
| bb["msp_analysis"] = payload["msp_analysis"] | |
| bb["context_bus_version"] = CONTEXT_BUS_VERSION | |
| return bb | |
| def _is_known(field: str, company: dict, gaps: list) -> bool: | |
| company = company or {} | |
| if field == "pkd": | |
| pkd = company.get("pkd_codes") or company.get("pkd") or [] | |
| if isinstance(pkd, str): | |
| return bool(pkd.strip()) | |
| return bool(pkd) | |
| if field == "region": | |
| return bool(company.get("voivodeship") or company.get("region")) | |
| if field == "nip": | |
| return bool(company.get("nip")) | |
| if field == "description": | |
| industry = (company.get("industry") or "").strip() | |
| if len(industry) >= 20: | |
| return True | |
| for gap in gaps or []: | |
| if "opis" in str(gap).lower(): | |
| return False | |
| return bool(industry) | |
| return False | |
| def filter_clarifying_questions( | |
| questions: List[str], | |
| company: dict, | |
| gaps: list | None = None, | |
| ) -> List[str]: | |
| """Skip questions for fields already present in company profile.""" | |
| gaps = gaps or [] | |
| result: List[str] = [] | |
| for question in questions or []: | |
| if not question: | |
| continue | |
| q_lower = question.lower() | |
| skip = False | |
| if "pkd" in q_lower and _is_known("pkd", company, gaps): | |
| skip = True | |
| if ("wojew" in q_lower or "region" in q_lower) and _is_known("region", company, gaps): | |
| skip = True | |
| if "nip" in q_lower and _is_known("nip", company, gaps): | |
| skip = True | |
| if ( | |
| "opis" in q_lower | |
| or "cel projektu" in q_lower | |
| or "działalno" in q_lower | |
| or ("główn" in q_lower and "pkd" in q_lower) | |
| ) and _is_known("description", company, gaps): | |
| skip = True | |
| if not skip: | |
| result.append(question) | |
| return result | |