Spaces:
Running
Running
| """Document intel: Trafilatura HTML, PyMuPDF/pypdf, PL/EU legal citations.""" | |
| from __future__ import annotations | |
| from pathlib import Path | |
| import pytest | |
| SAMPLE_HTML = """ | |
| <!DOCTYPE html> | |
| <html><head><title>Regulamin naboru SMART</title></head> | |
| <body> | |
| <nav>Menu · Kontakt · Logowanie · Cookie banner</nav> | |
| <main> | |
| <article> | |
| <h1>Regulamin naboru Ścieżka SMART</h1> | |
| <p>Niniejszy regulamin określa zasady ubiegania się o dofinansowanie | |
| zgodnie z rozporządzeniem (UE) 2021/1058 oraz CELEX 32021R1058.</p> | |
| <p>Podstawa prawna: Dz.U. z 2021 r. poz. 1234. Termin składania wniosków | |
| do 30.09.2027 r.</p> | |
| <p>Art. 14 ust. 1 określa obowiązki beneficjenta w zakresie kwalifikowalności.</p> | |
| <table><tr><th>Kryterium</th><th>Punkty</th></tr> | |
| <tr><td>Innowacyjność</td><td>30</td></tr></table> | |
| </article> | |
| </main> | |
| <footer>Copyright PARP · Polityka prywatności · Mapa strony</footer> | |
| </body></html> | |
| """ | |
| def test_html_main_content_strips_chrome(): | |
| from core.document_intel.html_extract import html_to_clean_text, extract_main_content | |
| result = html_to_clean_text(SAMPLE_HTML, url="https://www.parp.gov.pl/smart") | |
| text = result["text"] | |
| assert result["chars"] >= 80 | |
| assert "Regulamin naboru" in text or "SMART" in text | |
| assert "Cookie banner" not in text or result["extractor"] in ("trafilatura", "bs4") | |
| # main content must keep legal bits | |
| assert "2021" in text | |
| assert extract_main_content(SAMPLE_HTML) | |
| def test_legal_citations_celex_du_ue(): | |
| from core.document_intel.legal_citations import extract_legal_citations | |
| text = ( | |
| "Zgodnie z CELEX:32021R1058 oraz rozporządzeniem (UE) 2021/1058 " | |
| "oraz Dz.U. z 2021 r. poz. 1234 Art. 14 ust. 1." | |
| ) | |
| out = extract_legal_citations(text) | |
| assert out["count"] >= 2 | |
| assert any("32021R1058" in c for c in out["celex_ids"]) | |
| assert any("DU/2021/1234" == c for c in out["du_refs"]) | |
| assert out["ue_regs"] | |
| # no invent | |
| empty = extract_legal_citations("Brak podstawy prawnej w tym akapicie.") | |
| assert empty["count"] == 0 or not empty["celex_ids"] | |
| def test_extract_from_html_pipeline_includes_legal(): | |
| from core.document_intel.pipeline import extract_from_html | |
| out = extract_from_html(SAMPLE_HTML, url="https://example.gov.pl/reg") | |
| assert out["ok"] is True | |
| assert out["legal"]["count"] >= 1 | |
| assert out["extractor"] in ("trafilatura", "bs4") | |
| def test_pdf_extract_from_simple_pdf(tmp_path: Path): | |
| """Generate minimal PDF via reportlab or skip if unavailable; prefer pymupdf write.""" | |
| pdf_path = tmp_path / "reg.pdf" | |
| try: | |
| import fitz | |
| doc = fitz.open() | |
| page = doc.new_page() | |
| page.insert_text( | |
| (72, 72), | |
| "Regulamin testowy. CELEX 32021R0695. Dz.U. z 2022 r. poz. 55.", | |
| ) | |
| doc.save(str(pdf_path)) | |
| doc.close() | |
| except Exception: | |
| pytest.skip("pymupdf cannot create PDF in this env") | |
| from core.document_intel.pdf_extract import extract_pdf_text | |
| from core.document_intel.legal_citations import extract_legal_citations | |
| result = extract_pdf_text(pdf_path) | |
| assert result["parser"] in ("pymupdf", "pypdf") | |
| assert result["chars"] >= 20 | |
| assert "Regulamin" in result["text"] or "CELEX" in result["text"] | |
| cites = extract_legal_citations(result["text"]) | |
| assert cites["count"] >= 1 | |
| def test_pdf_parser_local_cascade_uses_document_intel(tmp_path: Path): | |
| try: | |
| import fitz | |
| except ImportError: | |
| pytest.skip("pymupdf missing") | |
| pdf_path = tmp_path / "local.pdf" | |
| doc = fitz.open() | |
| page = doc.new_page() | |
| page.insert_text((72, 72), "Dokument grantowy PARP SMART § 1. Podstawa CELEX 32021R1058.") | |
| doc.save(str(pdf_path)) | |
| doc.close() | |
| from rag_pipeline.pdf_parser import _parse_local_pdf_sync | |
| out = _parse_local_pdf_sync(str(pdf_path)) | |
| assert out["text"] | |
| assert out["parser"] in ("pymupdf", "pypdf") | |
| assert "SMART" in out["text"] or "CELEX" in out["text"] or "PARP" in out["text"] | |
| def test_eurlex_batch_still_rejects_free_text_titles(): | |
| """Regression: document intel must not weaken EUR-Lex legal-ID gate.""" | |
| from integrations.eurlex_client import _sanitize_search_query, is_valid_eurlex_query | |
| assert _sanitize_search_query("PARP Harmonogram naborów SMART") == "" | |
| assert is_valid_eurlex_query("32021R1058") | |
| def test_stealth_fetch_module_and_domain_gate(): | |
| from core.document_intel.stealth_fetch import ( | |
| is_stealth_fetch_enabled, | |
| should_use_stealth, | |
| stealth_get, | |
| ) | |
| assert is_stealth_fetch_enabled() is True | |
| assert should_use_stealth("https://www.parp.gov.pl/component/grants") | |
| assert should_use_stealth("https://www.bgk.pl/oferta/") | |
| assert not should_use_stealth("https://example.com/page") | |
| # Offline: invalid URL | |
| bad = stealth_get("not-a-url") | |
| assert bad["ok"] is False | |
| # Mocked success path — no network | |
| from unittest.mock import MagicMock, patch | |
| class FakeResp: | |
| status_code = 200 | |
| text = "<html><body><main>" + ("Regulamin SMART " * 40) + "</main></body></html>" | |
| content = text.encode() | |
| headers = {"content-type": "text/html"} | |
| url = "https://www.parp.gov.pl/x" | |
| fake_mod = MagicMock() | |
| fake_mod.get = MagicMock(return_value=FakeResp()) | |
| with patch.dict("sys.modules", {"curl_cffi": MagicMock(), "curl_cffi.requests": fake_mod}): | |
| # re-import path uses from curl_cffi import requests | |
| with patch("core.document_intel.stealth_fetch.stealth_get") as direct: | |
| # unit the real function with patched import inside | |
| pass | |
| # Call real implementation with patched curl_cffi.requests | |
| import core.document_intel.stealth_fetch as sf | |
| with patch.object(sf, "is_stealth_fetch_enabled", return_value=True): | |
| with patch( | |
| "curl_cffi.requests.get", | |
| return_value=FakeResp(), | |
| ): | |
| # stealth_get does `from curl_cffi import requests as curl_requests` | |
| import curl_cffi.requests as cr | |
| with patch.object(cr, "get", return_value=FakeResp()): | |
| out = stealth_get("https://www.parp.gov.pl/demo", timeout=5) | |
| assert out["ok"] is True | |
| assert out["status_code"] == 200 | |
| assert len(out["text"]) >= 200 | |
| assert out["impersonate"] | |
| def test_crawl4ai_hard_blocked_uses_stealth_not_empty(monkeypatch): | |
| """PARP URL must try stealth instead of hard-skipping to empty.""" | |
| import asyncio | |
| from unittest.mock import AsyncMock, patch | |
| from core.crawl4ai_client import scrape_url_to_markdown | |
| async def fake_stealth(url, **kwargs): | |
| return { | |
| "ok": True, | |
| "status_code": 200, | |
| "text": ( | |
| "<html><body><main><h1>PARP SMART</h1>" | |
| + ("<p>Treść regulaminu naboru. " * 30) | |
| + "</main></body></html>" | |
| ), | |
| "content": b"", | |
| "headers": {}, | |
| "final_url": url, | |
| "impersonate": "chrome", | |
| "error": None, | |
| } | |
| async def _run(): | |
| with patch( | |
| "core.document_intel.stealth_fetch.stealth_get_async", | |
| new=AsyncMock(side_effect=fake_stealth), | |
| ): | |
| md = await scrape_url_to_markdown("https://www.parp.gov.pl/component/grants/grants") | |
| assert md | |
| assert "SMART" in md or "regulaminu" in md.lower() or "naboru" in md.lower() | |
| assert len(md) >= 80 | |
| asyncio.run(_run()) | |
| def test_live_stealth_parp_optional(): | |
| """Optional live check — skip if network/WAF blocks CI.""" | |
| import os | |
| if os.environ.get("RUN_LIVE_STEALTH", "").lower() not in ("1", "true", "yes"): | |
| pytest.skip("Set RUN_LIVE_STEALTH=1 to hit PARP live") | |
| from core.document_intel.stealth_fetch import stealth_get | |
| out = stealth_get("https://www.parp.gov.pl/", timeout=25) | |
| assert out["ok"] is True | |
| assert out["status_code"] == 200 | |
| assert len(out.get("text") or "") > 1000 | |
| def test_page_fetcher_hash_uses_main_content_when_available(): | |
| """HttpPageFetcher should prefer trafilatura main-text hash when possible.""" | |
| import asyncio | |
| from unittest.mock import MagicMock, patch | |
| from core.grants.page_fetcher import HttpPageFetcher, content_hash | |
| from core.document_intel.html_extract import extract_main_content | |
| main = extract_main_content(SAMPLE_HTML) | |
| assert main | |
| fake_resp = MagicMock() | |
| fake_resp.text = SAMPLE_HTML | |
| fake_resp.status_code = 200 | |
| async def _run(): | |
| with patch("requests.get", return_value=fake_resp): | |
| fetcher = HttpPageFetcher(timeout=5) | |
| page = await fetcher.fetch("https://www.parp.gov.pl/demo") | |
| assert page.status_code == 200 | |
| assert page.body == SAMPLE_HTML | |
| # hash should match main content when trafilatura/bs4 works | |
| assert page.content_hash == content_hash(main) or page.content_hash == content_hash( | |
| SAMPLE_HTML | |
| ) | |
| assert page.source in ("http+trafilatura", "http") | |
| asyncio.run(_run()) | |