""" Anty-halucynacyjna normalizacja i ekstrakcja terminów naborów. Zasady: - Nigdy nie wstawiamy domyślnego 31.12 „na ślepo” bez kontekstu. - Preferujemy jawne pole termin_do / deadline. - Z tekstu wyciągamy daty tylko przy frazach naborowych (do / termin / nabór …). - Format wyjściowy zawsze YYYY-MM-DD lub pusty string. """ from __future__ import annotations import re from dataclasses import dataclass from datetime import date, datetime, timedelta from typing import Any, Dict, List, Optional, Sequence, Tuple PL_MONTHS = { "stycznia": 1, "styczeń": 1, "styczen": 1, "sty": 1, "lutego": 2, "luty": 2, "lut": 2, "marca": 3, "marzec": 3, "mar": 3, "kwietnia": 4, "kwiecień": 4, "kwiecien": 4, "kwi": 4, "maja": 5, "maj": 5, "czerwca": 6, "czerwiec": 6, "cze": 6, "lipca": 7, "lipiec": 7, "lip": 7, "sierpnia": 8, "sierpień": 8, "sierpien": 8, "sie": 8, "września": 9, "wrzesnia": 9, "wrzesień": 9, "wrzesien": 9, "wrz": 9, "października": 10, "pazdziernika": 10, "październik": 10, "pazdziernik": 10, "paź": 10, "paz": 10, "listopada": 11, "listopad": 11, "lis": 11, "grudnia": 12, "grudzień": 12, "grudzien": 12, "gru": 12, } # Prefer end-of-range when a range is present _RANGE_RE = re.compile( r"(\d{1,2}[./\-]\d{1,2}[./\-]\d{2,4}|\d{1,2}\s+\w+\s+\d{4}|\d{4}-\d{2}-\d{2})" r"\s*[-–—do]+\s*" r"(\d{1,2}[./\-]\d{1,2}[./\-]\d{2,4}|\d{1,2}\s+\w+\s+\d{4}|\d{4}-\d{2}-\d{2})", re.IGNORECASE, ) _EXPLICIT_PHRASE_RE = re.compile( r"(?:" r"termin(?:\s+naboru|\s+składania|\s+zgłoszeń|\s+wniosków)?|" r"nabór\s+(?:trwa\s+)?do|" r"nabor\s+(?:trwa\s+)?do|" r"składanie\s+wniosków\s+do|" r"wnioski\s+do|" r"do\s+dnia|" r"zamknięcie\s+naboru|" r"deadline|" r"do\b" r")" r"\s*[:\-]?\s*" r"(" r"\d{4}-\d{2}-\d{2}" r"|\d{1,2}[./\-]\d{1,2}[./\-]\d{2,4}" r"|\d{1,2}\s+(?:" + "|".join(re.escape(k) for k in sorted(PL_MONTHS, key=len, reverse=True)) + r")\s+\d{4}" r")", re.IGNORECASE, ) _PL_WORD_DATE_RE = re.compile( r"\b(\d{1,2})\s+(" + "|".join(re.escape(k) for k in sorted(PL_MONTHS, key=len, reverse=True)) + r")\s+(\d{4})\b", re.IGNORECASE, ) _ISO_RE = re.compile(r"\b(20\d{2})-(\d{2})-(\d{2})\b") _EU_DOT_RE = re.compile(r"\b(\d{1,2})[./\-](\d{1,2})[./\-](\d{2,4})\b") _QX_RE = re.compile(r"\bQ([1-4])\s*(20\d{2})\b", re.IGNORECASE) # Fields considered primary (high confidence) PRIMARY_DEADLINE_KEYS = ( "termin_do", "deadline", "data_zakonczenia", "data_zakończenia", "data_konca", "data_końca", "end_date", "closing_date", "application_deadline", "termin_skladania", "termin_składania", "termin_naboru", ) SECONDARY_DEADLINE_KEYS = ( "termin", "termin_od", "start_date", "data_rozpoczecia", "data_rozpoczęcia", ) TEXT_KEYS = ( "description", "opis_krotki", "name", "nazwa", "eligibility_criteria", "warunki_wejscia", "program_goals", "beneficjenci", "status", ) @dataclass(frozen=True) class DeadlineHit: iso: str # YYYY-MM-DD confidence: float # 0..1 source: str # field name or "text:phrase" raw: str = "" def as_dict(self) -> Dict[str, Any]: return { "deadline": self.iso, "confidence_on_dates": self.confidence, "deadline_source": self.source, "deadline_raw": self.raw, } def _safe_date(year: int, month: int, day: int) -> Optional[date]: try: if month == 2: day = min(day, 29 if year % 4 == 0 and (year % 100 != 0 or year % 400 == 0) else 28) elif month in (4, 6, 9, 11): day = min(day, 30) else: day = min(day, 31) if not (1 <= month <= 12) or day < 1: return None if year < 2000 or year > 2040: return None return date(year, month, day) except (ValueError, TypeError): return None def _year_norm(y: str | int) -> int: yi = int(y) if yi < 100: return 2000 + yi if yi < 50 else 1900 + yi return yi def _month_from_token(mon: str) -> Optional[int]: mon_l = mon.strip().lower().replace("ę", "e").replace("ą", "a").replace("ś", "s").replace("ć", "c").replace("ń", "n").replace("ó", "o").replace("ź", "z").replace("ż", "z") # try original first if mon.isdigit(): m = int(mon) return m if 1 <= m <= 12 else None mon_raw = mon.strip().lower() if mon_raw in PL_MONTHS: return PL_MONTHS[mon_raw] for k, v in PL_MONTHS.items(): if mon_raw.startswith(k[:3]) or mon_l.startswith(k[:3].replace("ę", "e")): return v # ascii-folded keys already in map partially folded = { "stycznia": 1, "styczen": 1, "lutego": 2, "marca": 3, "kwietnia": 4, "maja": 5, "czerwca": 6, "lipca": 7, "sierpnia": 8, "wrzesnia": 9, "pazdziernika": 10, "listopada": 11, "grudnia": 12, } return folded.get(mon_l) def parse_deadline_robust(deadline_raw: str, context_text: str = "") -> Optional[date]: """ Robust parser for a single deadline candidate string. Supports ISO, EU dots/slashes, Polish month names, Qx YYYY, ranges (end date). """ if deadline_raw and str(deadline_raw).strip(): dl = str(deadline_raw).strip() dl_l = dl.lower() # Continuous / open — no concrete date if any( k in dl_l for k in ( "ciągły", "ciagly", "continuous", "beztermin", "n/a", "brak", "open-ended", "cały rok", "caly rok", ) ): return None # Truncated years like 19.06.202 (3-digit year) — refuse if re.search(r"\b\d{1,2}[./]\d{1,2}[./]\d{3}\b", dl) and not re.search( r"\b\d{1,2}[./]\d{1,2}[./]\d{4}\b", dl ): return None # Prefer full ISO (and ISO datetime) BEFORE loose EU patterns — avoids # matching "26-07-01" inside "2026-07-01T23:59:59" as day=26 month=07 year=01. m_iso = _ISO_RE.search(dl) if m_iso: y, mo, d = m_iso.groups() parsed = _safe_date(int(y), int(mo), int(d)) if parsed: return parsed # ISO datetime prefix YYYY-MM-DDTHH:MM… m_iso_dt = re.search(r"\b(20\d{2})-(\d{2})-(\d{2})[T\s]", dl) if m_iso_dt: y, mo, d = m_iso_dt.groups() parsed = _safe_date(int(y), int(mo), int(d)) if parsed: return parsed # Year-Month only → last day of month (e.g. 2026-03, 03.2026) ym = re.fullmatch(r"(20\d{2})[-/.](\d{1,2})", dl.strip()) if ym: year, month = int(ym.group(1)), int(ym.group(2)) if 1 <= month <= 12: if month == 12: return date(year, 12, 31) return date(year, month + 1, 1) - timedelta(days=1) # Month.Year only → last day of month (e.g. 12.2026) my = re.fullmatch(r"(\d{1,2})[./](20\d{2})", dl.strip()) if my: month, year = int(my.group(1)), int(my.group(2)) if 1 <= month <= 12: # last calendar day if month == 12: return date(year, 12, 31) return date(year, month + 1, 1).fromordinal( date(year, month + 1, 1).toordinal() - 1 ) # Range first → end date rm = _RANGE_RE.search(dl) if rm: end = parse_deadline_robust(rm.group(2), "") if end: return end # Direct strptime formats candidates = [dl, dl[:10], dl.split()[0] if " " in dl else dl] for fmt in ( "%Y-%m-%d", "%d.%m.%Y", "%d-%m-%Y", "%d/%m/%Y", "%Y.%m.%d", "%d.%m.%y", "%d/%m/%y", "%d-%m-%y", ): for candidate in candidates: try: return datetime.strptime(candidate.strip(), fmt).date() except Exception: continue # Polish "15 maja 2026" / EN "15 March 2026" m = _PL_WORD_DATE_RE.search(dl) if m: day, mon, year = m.groups() month = _month_from_token(mon) if month: d = _safe_date(_year_norm(year), month, int(day)) if d: return d m_en = re.search( r"\b(\d{1,2})\s+(january|february|march|april|may|june|july|august|september|october|november|december)\s+(20\d{2})\b", dl_l, ) if m_en: day_i = int(m_en.group(1)) mon_map = { "january": 1, "february": 2, "march": 3, "april": 4, "may": 5, "june": 6, "july": 7, "august": 8, "september": 9, "october": 10, "november": 11, "december": 12, } parsed = _safe_date(int(m_en.group(3)), mon_map[m_en.group(2)], day_i) if parsed: return parsed # "do 30.06.2026" / "30 czerwca 2026" loose — only if day token is 1–31 and # not a 4-digit year fragment (block matching inside YYYY-MM-DD leftovers). m = re.search( r"(?:do|termin|naboru|do\s+dnia|deadline|closes?)?\s*[:\-]?\s*(\d{1,2})[.\s/\-]+(\d{1,2}|\w+)[.\s/\-]+(\d{2,4})", dl_l, re.IGNORECASE, ) if m: d, mon, y = m.groups() day_i = int(d) # Reject year-leading fragments (e.g. 20 from 2026) if day_i <= 31 and not (len(d) == 2 and day_i >= 20 and str(y).isdigit() and int(y) <= 31): month = _month_from_token(mon) if month: parsed = _safe_date(_year_norm(y), month, day_i) if parsed: return parsed m = _EU_DOT_RE.search(dl) if m: d, mo, y = m.groups() # guard: day/month swap if day>12 already EU style day_i, mon_i = int(d), int(mo) if mon_i > 12 and day_i <= 12: day_i, mon_i = mon_i, day_i # Refuse absurd year fragments from partial ISO leftovers yi = _year_norm(y) if 2000 <= yi <= 2040 and 1 <= mon_i <= 12 and 1 <= day_i <= 31: parsed = _safe_date(yi, mon_i, day_i) if parsed: return parsed qm = _QX_RE.search(dl) if qm: q, y = int(qm.group(1)), int(qm.group(2)) month = {1: 3, 2: 6, 3: 9, 4: 12}[q] return date(y, month, 15) # Year-only with deadline wording → Dec 31 if re.search(r"(do|termin|nabor)", dl_l) and re.search(r"\b(20[2-3]\d)\b", dl_l): if not re.search(r"\d{1,2}[./]\d{1,2}", dl_l): y = int(re.search(r"\b(20[2-3]\d)\b", dl_l).group(1)) return date(y, 12, 31) ctx = (context_text or "").strip() if ctx: hit = extract_deadline_from_text(ctx) if hit: try: return datetime.strptime(hit.iso, "%Y-%m-%d").date() except Exception: return None return None def extract_deadline_from_text(text: str) -> Optional[DeadlineHit]: """Extract deadline from free text using explicit nabor phrases first.""" if not (text or "").strip(): return None t = text lower = t.lower() # Prefer "do X" / "termin …" phrases for m in _EXPLICIT_PHRASE_RE.finditer(t): raw = m.group(1) parsed = parse_deadline_robust(raw, "") if parsed: conf = 0.82 if "termin" in m.group(0).lower() or "nabór" in m.group(0).lower() else 0.72 return DeadlineHit(parsed.isoformat(), conf, "text:phrase", raw) # Range in text → end date rm = _RANGE_RE.search(t) if rm: end = parse_deadline_robust(rm.group(2), "") if end: return DeadlineHit(end.isoformat(), 0.75, "text:range", rm.group(0)) # Horizon / EU style: "2026-05-28 2026-08-27 Open" → take last ISO as end iso_dates = list(_ISO_RE.finditer(t)) if len(iso_dates) >= 2 and any(k in lower for k in ("open", "horizon", "call", "grant", "nabór", "nabor")): y, mo, d = iso_dates[-1].groups() parsed = _safe_date(int(y), int(mo), int(d)) if parsed: return DeadlineHit(parsed.isoformat(), 0.7, "text:iso_range_end", iso_dates[-1].group(0)) # Polish word date near nabor keywords (or title-prefix news-date for nabor pages) naborish = any( k in lower for k in ( "termin", "nabór", "nabor", "wniosk", "składan", "deadline", "harmonogram", "dofinans", "grant", "konkurs", "call", ) ) pl_hits = list(_PL_WORD_DATE_RE.finditer(t)) for m in pl_hits: window = lower[max(0, m.start() - 40) : m.end() + 15] near = any( k in window for k in ("termin", "do ", "nabór", "nabor", "wniosk", "składan", "deadline", "harmonogram") ) if near or (naborish and m.start() < 40): day, mon, year = m.groups() month = _month_from_token(mon) if month: d = _safe_date(_year_norm(year), month, int(day)) if d: conf = 0.72 if near else 0.58 return DeadlineHit(d.isoformat(), conf, "text:pl_month", m.group(0)) # EU numeric dates near keywords for m in _EU_DOT_RE.finditer(t): window = lower[max(0, m.start() - 35) : m.end() + 10] if any(k in window for k in ("termin", "do ", "nabór", "nabor", "wniosk", "deadline", "do\u00a0")): d, mo, y = m.groups() day_i, mon_i = int(d), int(mo) if mon_i > 12 and day_i <= 12: day_i, mon_i = mon_i, day_i parsed = _safe_date(_year_norm(y), mon_i, day_i) if parsed: return DeadlineHit(parsed.isoformat(), 0.7, "text:eu_date", m.group(0)) # Single clear future date in short nabor-ish blurb (last resort, low confidence) if naborish: candidates: List[date] = [] raws: List[str] = [] for m in list(_ISO_RE.finditer(t)) + list(_EU_DOT_RE.finditer(t)) + list(_PL_WORD_DATE_RE.finditer(t)): parsed = parse_deadline_robust(m.group(0), "") if parsed and parsed.year >= date.today().year - 1: candidates.append(parsed) raws.append(m.group(0)) if candidates: # Prefer latest date as closing (common in "open from … to …") best_i = max(range(len(candidates)), key=lambda i: candidates[i]) return DeadlineHit( candidates[best_i].isoformat(), 0.55, "text:fallback_latest", raws[best_i], ) return None def extract_grant_deadline( record: Dict[str, Any], *, prefer_end_of_range: bool = True, allow_text_fallback: bool = True, ) -> Optional[DeadlineHit]: """ Unified deadline extraction for ingest/mapper/post-import. Order: 1. Primary fields (termin_do, deadline, …) 2. Secondary (termin, termin_od only if range end elsewhere) 3. Free-text extraction with nabor phrases """ if not record: return None # Continuous enrollment → no concrete deadline status = str(record.get("status") or "").lower() blob_early = " ".join( str(record.get(k) or "") for k in ("status", "name", "nazwa", "description", "opis_krotki") ).lower() if any(k in status or k in blob_early for k in ("ciągły", "ciagly", "beztermin", "continuous")): # still allow explicit termin_do if present pass # 1) Primary fields for key in PRIMARY_DEADLINE_KEYS: raw = record.get(key) if raw is None or str(raw).strip() == "": continue raw_s = str(raw).strip() # Range in field → end if prefer_end_of_range: rm = _RANGE_RE.search(raw_s) if rm: end = parse_deadline_robust(rm.group(2), "") if end: return DeadlineHit(end.isoformat(), 0.9, key + ":range_end", raw_s) parsed = parse_deadline_robust(raw_s, "") if parsed: conf = 0.92 if key in ("termin_do", "deadline", "application_deadline") else 0.85 # year-only from bare field is weaker if re.fullmatch(r"20\d{2}", raw_s.strip()): conf = 0.45 return DeadlineHit(parsed.isoformat(), conf, key, raw_s) # 2) Secondary — only if looks like end date or ISO for key in SECONDARY_DEADLINE_KEYS: raw = record.get(key) if not raw: continue raw_s = str(raw).strip() if key == "termin_od": # start date alone is not deadline; skip unless paired later continue parsed = parse_deadline_robust(raw_s, "") if parsed: return DeadlineHit(parsed.isoformat(), 0.65, key, raw_s) # 3) Text blobs if allow_text_fallback: parts: List[str] = [] for key in TEXT_KEYS: val = record.get(key) if val: parts.append(str(val)) # also nested raw_data raw_data = record.get("raw_data") if isinstance(record.get("raw_data"), dict) else {} for key in TEXT_KEYS + PRIMARY_DEADLINE_KEYS: val = raw_data.get(key) if raw_data else None if val: parts.append(str(val)) text = "\n".join(parts) hit = extract_deadline_from_text(text) if hit: return hit return None def deadline_to_iso(value: Any, context: str = "") -> str: """Convenience: any value → YYYY-MM-DD or ''.""" if value is None: return "" if isinstance(value, date) and not isinstance(value, datetime): return value.isoformat() if isinstance(value, datetime): return value.date().isoformat() s = str(value).strip() if not s: return "" parsed = parse_deadline_robust(s, context) return parsed.isoformat() if parsed else "" def apply_deadline_to_grant_dict(grant: Dict[str, Any]) -> Dict[str, Any]: """ Mutates a copy of grant dict with deadline fields filled from multi-source extract. Safe for continuous enrollments (leaves empty). """ g = dict(grant) status_l = str(g.get("status") or "").lower() continuous = any(k in status_l for k in ("ciągły", "ciagly", "continuous", "beztermin")) existing = str(g.get("deadline") or "").strip() existing_parsed = parse_deadline_robust(existing, "") if existing else None if existing_parsed: g["deadline"] = existing_parsed.isoformat() g.setdefault("confidence_on_dates", 0.9) g.setdefault("deadline_source", "existing") return g # Build extract record from grant + common PL aliases record = dict(g) raw = g.get("raw_data") if isinstance(g.get("raw_data"), dict) else {} for k, v in raw.items(): record.setdefault(k, v) hit = extract_grant_deadline(record, allow_text_fallback=not continuous) if hit: g["deadline"] = hit.iso g["confidence_on_dates"] = hit.confidence g["deadline_source"] = hit.source g["deadline_raw"] = hit.raw elif continuous: g["deadline"] = "" g["confidence_on_dates"] = 0.35 g["deadline_source"] = "continuous" elif not existing: g["deadline"] = "" g.setdefault("confidence_on_dates", 0.25) g.setdefault("deadline_source", "missing") return g def normalize_grant_date_status( grant: Dict[str, Any], raw_context: str = "", is_current_only: bool = False, ) -> Dict[str, Any]: g = apply_deadline_to_grant_dict(dict(grant)) text = (raw_context or str(g.get("description", "")) + " " + str(g.get("name", ""))).lower() name = (g.get("name") or "").lower() deadline_raw = (g.get("deadline") or "").strip() status_raw = (g.get("status") or "").strip().lower() g["source_status_raw"] = status_raw or "unknown" g.setdefault("confidence_on_dates", 0.6) today = date.today() closed_kw = [ "upłynął", "nabór zakończony", "zakończony nabór", "archiwum", "nie przyjmujemy już", "zamknięty nabór", "termin upłynął", ] if any(kw in text or kw in name for kw in closed_kw): g.update( status="closed", normalized_status="closed", is_outdated_warning=True, is_current_only=False, confidence_on_dates=0.92, ) return g continuous_kw = ("ciągły", "ciagly", "continuous", "bezterminow") if status_raw in continuous_kw or any(k in text for k in continuous_kw): g.update( normalized_status="active", status="active", deadline="", confidence_on_dates=0.35, is_current_only=True, requires_manual_verification=True, deadline_source="continuous", ) return g is_fin = g.get("instrument_type") in [ "loan", "guarantee", "interest_subsidy", "other_financial", "mixed", ] if is_fin and ( not deadline_raw or "ciągły" in deadline_raw.lower() or "n/a" in deadline_raw.lower() ): g.update( normalized_status="active", status="active", confidence_on_dates=0.85, is_current_only=True, ) return g parsed_deadline = parse_deadline_robust(deadline_raw, text + " " + name) if parsed_deadline: g["deadline"] = parsed_deadline.isoformat() g["confidence_on_dates"] = max(float(g.get("confidence_on_dates") or 0), 0.88 if parsed_deadline.year >= 2025 else 0.75) if parsed_deadline < today: g.update( normalized_status="closed", status="closed", is_outdated_warning=True, is_current_only=False, ) return g elif any(kw in text for kw in ["upłyn", "zakończ", "archiw"]) or status_raw == "closed": g.update( normalized_status="closed", status="closed", is_outdated_warning=True, is_current_only=False, confidence_on_dates=0.7, ) return g planned_kw = ["planowany", "rusza od", "wkrótce", "w przygotowaniu", "potrwa od"] if any(kw in text for kw in planned_kw) and (not parsed_deadline or parsed_deadline > today): g.update( normalized_status="planned", status="planned", confidence_on_dates=max(float(g.get("confidence_on_dates") or 0), 0.74), is_current_only=False, ) if is_current_only: g["is_outdated_warning"] = True return g final_status = g.get("status", "active") or "active" g["normalized_status"] = final_status g["status"] = final_status is_current = (final_status in ("active", "planned")) and not g.get("is_outdated_warning", False) g["is_current_only"] = bool(is_current) if is_current_only and not is_current: g.update(is_outdated_warning=True, normalized_status="closed", status="closed") if not deadline_raw and float(g.get("confidence_on_dates") or 0) >= 0.6: g["confidence_on_dates"] = 0.3 return g def backfill_deadlines_report( records: Sequence[Dict[str, Any]], ) -> Dict[str, Any]: """Stats helper for tests/ops — how many deadlines we can fill.""" total = len(records) before_raw = sum(1 for r in records if str(r.get("deadline") or r.get("termin_do") or "").strip()) before_parseable = 0 for r in records: raw = str(r.get("deadline") or r.get("termin_do") or "").strip() if raw and parse_deadline_robust(raw, ""): before_parseable += 1 filled = 0 improved = 0 sources: Dict[str, int] = {} for r in records: raw = str(r.get("deadline") or r.get("termin_do") or "").strip() had_parseable = bool(raw and parse_deadline_robust(raw, "")) hit = extract_grant_deadline(r) if hit: filled += 1 src_key = hit.source.split(":")[0] sources[src_key] = sources.get(src_key, 0) + 1 if not had_parseable: improved += 1 return { "total": total, "with_deadline_before": before_raw, "with_deadline_before_parseable": before_parseable, "with_deadline_after": filled, "newly_filled": improved, "pct_before": round(100 * before_raw / max(total, 1), 1), "pct_before_parseable": round(100 * before_parseable / max(total, 1), 1), "pct_after": round(100 * filled / max(total, 1), 1), "by_source": sources, }