Spaces:
Running
Running
| """Ocena kompletności danych programu (Warstwa 1 — Curated KB).""" | |
| from __future__ import annotations | |
| from typing import Any, Dict, List | |
| from core.grants.regulation_url_quality import ( | |
| collect_all_regulation_urls, | |
| collect_regulation_url_candidates, | |
| is_valid_regulation_url, | |
| pick_best_regulation_url, | |
| ) | |
| REQUIRED_FIELDS = ( | |
| "name", | |
| "description", | |
| "eligible_beneficiaries", | |
| "eligible_company_sizes", | |
| "eligible_regions", | |
| "eligible_pkd", | |
| "regulation_url", | |
| "official_page_url", | |
| "program_year", | |
| "operator", | |
| "instrument_type", | |
| "status", | |
| "deadline", | |
| "budget_max_pln", | |
| ) | |
| LINK_FIELDS = ("regulation_url", "official_page_url", "application_url", "precise_regulation_url", "url") | |
| def _has_value(val: Any) -> bool: | |
| if val is None: | |
| return False | |
| if isinstance(val, (list, dict)): | |
| return len(val) > 0 | |
| if isinstance(val, (int, float)): | |
| return val > 0 | |
| return bool(str(val).strip()) | |
| def _resolved_regulation_url(row, raw: Dict[str, Any]) -> str: | |
| merged = dict(raw) | |
| merged["regulation_url"] = row.regulation_url or raw.get("regulation_url") | |
| merged["precise_regulation_url"] = row.precise_regulation_url or raw.get("precise_regulation_url") | |
| return pick_best_regulation_url(collect_regulation_url_candidates(merged)) | |
| def grant_dict_from_row(row) -> Dict[str, Any]: | |
| """Łączy kolumny ORM z raw_data — kolumny DB mają pierwszeństwo (live research).""" | |
| raw = dict(row.raw_data or {}) | |
| raw["id"] = row.source_id | |
| raw["name"] = row.name | |
| raw["status"] = row.status | |
| raw["description"] = row.description or raw.get("description") or "" | |
| raw["operator"] = row.operator or row.program or raw.get("operator") or "" | |
| raw["program_year"] = row.program_year if row.program_year is not None else raw.get("program_year") | |
| raw["catalog_hidden"] = row.catalog_hidden | |
| raw["instrument_type"] = row.instrument_type or raw.get("instrument_type") or "grant" | |
| raw["eligible_regions"] = row.eligible_regions or raw.get("eligible_regions") or [] | |
| raw["eligible_company_sizes"] = row.eligible_company_sizes or raw.get("eligible_company_sizes") or [] | |
| raw["eligible_beneficiaries"] = row.eligible_beneficiaries or raw.get("beneficiary_tags") or [] | |
| raw["beneficiary_tags"] = raw.get("beneficiary_tags") or raw["eligible_beneficiaries"] | |
| raw["eligible_pkd"] = row.eligible_pkd or raw.get("eligible_pkd") or [] | |
| raw["deadline"] = row.deadline or raw.get("deadline") or "" | |
| raw["url"] = row.url or raw.get("url") or "" | |
| reg = _resolved_regulation_url(row, raw) | |
| raw["regulation_url"] = reg | |
| raw["precise_regulation_url"] = reg | |
| raw["regulation_urls"] = raw.get("regulation_urls") or collect_all_regulation_urls(raw) | |
| for link_field in ("link_regulamin", "zrodlo_prawdy", "legal_source"): | |
| link_val = str(raw.get(link_field) or "").strip() | |
| if link_val.startswith("http") and not is_valid_regulation_url(link_val): | |
| raw[link_field] = "" | |
| elif link_field == "legal_source" and reg: | |
| raw[link_field] = reg | |
| raw["official_page_url"] = row.official_page_url or row.url or raw.get("official_page_url") or "" | |
| raw["application_url"] = row.application_url or raw.get("link_ogloszenie") or raw.get("application_url") or "" | |
| raw.setdefault("budget_max_pln", row.budget_max_pln or raw.get("max_dofinansowanie_pln")) | |
| raw.setdefault("budget_min_pln", row.budget_min_pln or raw.get("min_dofinansowanie_pln")) | |
| raw.setdefault("program_goals", row.program_goals or "") | |
| raw.setdefault("eligibility_criteria", row.eligibility_criteria or raw.get("warunki_wejscia") or "") | |
| raw.setdefault("completeness_score", row.completeness_score or 0) | |
| raw.setdefault("source_credibility_score", row.source_credibility_score or 0.5) | |
| return raw | |
| def compute_completeness(item: Dict[str, Any]) -> Dict[str, Any]: | |
| """Zwraca score 0-100 i listę brakujących pól.""" | |
| present: List[str] = [] | |
| missing: List[str] = [] | |
| for field in REQUIRED_FIELDS: | |
| val = item.get(field) | |
| if field == "eligible_beneficiaries" and not _has_value(val): | |
| val = item.get("beneficiary_tags") | |
| if field == "budget_max_pln" and not _has_value(val): | |
| val = item.get("max_dofinansowanie_pln") or item.get("kwota_max") | |
| if field == "regulation_url": | |
| val = pick_best_regulation_url(collect_regulation_url_candidates(item)) or None | |
| if _has_value(val): | |
| present.append(field) | |
| else: | |
| missing.append(field) | |
| working_links = sum( | |
| 1 | |
| for f in LINK_FIELDS | |
| if str(item.get(f) or "").startswith("http") | |
| ) | |
| link_score = min(20, working_links * 7) | |
| field_score = int(len(present) / max(len(REQUIRED_FIELDS), 1) * 80) | |
| score = min(100, field_score + link_score) | |
| return { | |
| "completeness_score": score, | |
| "fields_present": present, | |
| "fields_missing": missing, | |
| "working_links": working_links, | |
| } | |
| def catalog_completeness_stats(items: List[Dict[str, Any]]) -> Dict[str, Any]: | |
| if not items: | |
| return {"avg_completeness": 0, "high_quality_pct": 0, "with_regulation_link_pct": 0} | |
| scores = [compute_completeness(i)["completeness_score"] for i in items] | |
| with_reg = sum( | |
| 1 | |
| for i in items | |
| if is_valid_regulation_url(str(i.get("regulation_url") or i.get("precise_regulation_url") or "")) | |
| ) | |
| return { | |
| "avg_completeness": round(sum(scores) / len(scores), 1), | |
| "high_quality_pct": round(100 * sum(1 for s in scores if s >= 70) / len(scores), 1), | |
| "with_regulation_link_pct": round(100 * with_reg / len(items), 1), | |
| "sample_size": len(items), | |
| } | |