Spaces:
Running
Running
File size: 5,667 Bytes
3a3734f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 | """Project creation helpers — eligibility, enrichment, early MSP analysis."""
from __future__ import annotations
import logging
from fastapi import HTTPException
logger = logging.getLogger(__name__)
def validate_grant_eligibility(external_context: dict) -> None:
"""Hard-block tworzenia projektu dla zamkniętych / przestarzałych naborów."""
if "selected_grant" not in external_context:
return
grant = external_context["selected_grant"]
if grant.get("status") == "closed" or grant.get("is_outdated_warning"):
logger.warning(
"[Projects] Próba utworzenia wniosku dla wygasłego naboru: %s",
grant.get("name"),
)
raise HTTPException(
status_code=400,
detail=(
"Ten nabór został oznaczony jako ZAMKNIĘTY lub PRZESTARZAŁY. "
"Ze względów bezpieczeństwa nie można utworzyć projektu dla tego naboru."
),
)
async def enrich_company_context_at_creation(enriched_context: dict, description: str | None) -> dict:
"""Pełne pobranie GUS/KRS + enrichment SUDOP/gaps na starcie projektu."""
if "company_data" not in enriched_context:
return enriched_context
if not enriched_context["company_data"].get("nip"):
return enriched_context
nip = enriched_context["company_data"]["nip"]
try:
from tools.company_search import fetch_regon_data
from integrations.krs_client import KRSClient
full_data = fetch_regon_data(nip) or {}
full_data = {**full_data, "nip": nip}
krs = full_data.get("krs") or full_data.get("numerKRS")
if krs:
try:
odpis = KRSClient.get_odpis_aktualny(str(krs))
if odpis:
krs_relations = KRSClient.extract_graph_relations(odpis)
full_data["krs_full"] = krs_relations
full_data["krs_zarzad"] = krs_relations.get("zarzad", [])
full_data["krs_wspolnicy"] = krs_relations.get("wspolnicy", [])
logger.info("[Projects] Pobrano pełny odpis KRS dla projektu (KRS: %s)", krs)
except Exception as krs_e:
logger.warning("[Projects] Nie udało się pobrać odpisu KRS: %s", krs_e)
enriched_context["company_data"] = full_data
logger.info("[Projects] Wzbogacono dane firmy na starcie projektu (NIP: %s)", nip)
try:
from core.registry_enrichment import enrich_company_profile
enrichment = await enrich_company_profile(
nip,
project_description=description or "",
include_web_intel=False,
)
enriched_company = enrichment.get("company_data") or {}
for krs_key in ("krs_full", "krs_zarzad", "krs_wspolnicy"):
if krs_key in full_data and krs_key not in enriched_company:
enriched_company[krs_key] = full_data[krs_key]
enriched_context["company_data"] = {**full_data, **enriched_company}
if enrichment.get("gaps"):
enriched_context["gaps"] = enrichment["gaps"]
if enrichment.get("risk_hints"):
enriched_context["risk_hints"] = enrichment["risk_hints"]
if enrichment.get("sudop"):
enriched_context["sudop"] = enrichment["sudop"]
logger.info(
"[Projects] Enrichment SUDOP/gaps wykonany dla NIP %s (luki: %s)",
nip,
len(enrichment.get("gaps", [])),
)
except Exception as enrich_e:
logger.warning("[Projects] Pełny enrichment (SUDOP/gaps) nieudany: %s", enrich_e)
except Exception as gus_e:
logger.warning(
"[Projects] Błąd wzbogacania danych GUS/KRS dla nowo tworzonego projektu: %s",
gus_e,
)
return enriched_context
def apply_early_msp_analysis(enriched_context: dict) -> dict:
"""Wczesna analiza MSP/GraphRAG przy tworzeniu projektu."""
if "company_data" not in enriched_context:
return enriched_context
nip = enriched_context["company_data"].get("nip")
if not nip:
return enriched_context
try:
from core.graph_rag.sme_verifier import sme_verifier
declared = enriched_context.get("declared_sme_status", "mikro")
msp_result = sme_verifier.verify_sme_status(nip=nip, declared_status=declared)
enriched_context["msp_analysis"] = msp_result
logger.info("[Projects] Wczesna analiza MSP wykonana dla NIP %s przy tworzeniu projektu", nip)
is_mock = msp_result.get("using_real_data") is False or "mock" in str(
msp_result.get("msp_risk_note", "")
).lower()
enriched_context["credibility_flags"] = {
"msp_analysis_performed": True,
"msp_using_real_data": msp_result.get("using_real_data", False),
"msp_risk": bool(msp_result.get("msp_risk_note")),
"gus_enriched": bool(
enriched_context.get("company_data", {}).get("pkd")
),
"overall_trust_level": (
"high"
if msp_result.get("using_real_data")
else "medium"
if not is_mock
else "low-mock"
),
}
except Exception as msp_e:
logger.warning("[Projects] Nie udało się wykonać wczesnej analizy MSP: %s", msp_e)
enriched_context["credibility_flags"] = {
"msp_analysis_performed": False,
"overall_trust_level": "unknown",
}
return enriched_context
|