Add regression tests for recatpcha

Author: adelavega

ace/ingest.py

@@ -10,6 +10,29 @@
 
 logger = logging.getLogger(__name__)
 
+
+def _article_quality_score(article):
+    """Score article quality for same-PMID duplicate resolution.
+
+    Higher is better. Prioritizes records that actually contain extracted
+    activations and usable tables over placeholder/blocked pages.
+    """
+    if article is None:
+        return (-1, -1, -1, -1)
+
+    tables = getattr(article, "tables", []) or []
+    n_tables = len(tables)
+    n_activations = sum(len(getattr(t, "activations", []) or []) for t in tables)
+    missing_source = 1 if getattr(article, "missing_source", False) else 0
+
+    # Sort by:
+    # 1) Most activations
+    # 2) Most tables
+    # 3) Prefer non-missing-source parses
+    # 4) Presence of any table at all
+    return (n_activations, n_tables, -missing_source, int(n_tables > 0))
+
+
 def _process_file_with_source(args):
     """Helper function to read, validate, and identify source for a file."""
     f, source_configs = args
@@ -167,23 +190,44 @@ def add_articles(db, files, commit=True, table_dir=None, limit=None,
         for args in tqdm(parse_args, desc="Parsing articles"):
             parsed_articles.append(_parse_article(args))
 
-    # Add successfully parsed articles to database
+    # Add successfully parsed articles to database.
+    # When pmid_filenames=True we can see duplicate PMID files from different
+    # source folders (e.g., one blocked/challenge page + one valid page). Keep
+    # the best parsed candidate per PMID within this run.
     missing_sources = []
-    for i, (f, article) in enumerate(parsed_articles):
-        if article is None:
-            missing_sources.append(f)
-            continue
-            
+    if pmid_filenames:
+        best_by_pmid = {}
+        for f, article in parsed_articles:
+            pmid = path.splitext(path.basename(f))[0]
+            if article is None:
+                missing_sources.append(f)
+                continue
+
+            score = _article_quality_score(article)
+            existing = best_by_pmid.get(pmid)
+            if existing is None or score > existing[2]:
+                best_by_pmid[pmid] = (f, article, score)
+
+        selected_articles = [(f, article) for (f, article, _) in best_by_pmid.values()]
+    else:
+        selected_articles = []
+        for f, article in parsed_articles:
+            if article is None:
+                missing_sources.append(f)
+                continue
+            selected_articles.append((f, article))
+
+    for i, (f, article) in enumerate(selected_articles):
         if get_config('SAVE_ARTICLES_WITHOUT_ACTIVATIONS') or article.tables:
             pmid = path.splitext(path.basename(f))[0] if pmid_filenames else None
             if pmid and db.article_exists(pmid):
                 if get_config('OVERWRITE_EXISTING_ROWS'):
                     db.delete_article(pmid)
                 else:
                     continue
-                    
+
             db.add(article)
-            if commit and (i % 100 == 0 or i == len(parsed_articles) - 1):
+            if commit and (i % 100 == 0 or i == len(selected_articles) - 1):
                 db.save()
 
     db.save()

ace/scrape.py

@@ -225,6 +225,10 @@ def _validate_scrape(html):
     'This site can’t be reached',
     'used Cloudflare to restrict access',
     '502 Bad Gateway',
+    'Checking your browser before accessing',
+    'Checking your browser - reCAPTCHA',
+    '/recaptcha/challengepage/',
+    'g-recaptcha',
     ]
 
     for pattern in patterns:

ace/tests/test_ace.py

@@ -476,3 +476,72 @@ def test_unknown_source_coordinate_table_with_force_ingest(test_weird_data_path,
     assert len(db_force.articles) >= 1
     assert len(db_force.articles[0].tables) >= 1
     assert _count_valid_activations(db_force.articles[0].tables) >= 1
+
+
+def test_ingest_prefers_best_duplicate_pmid_file(test_weird_data_path, tmp_path):
+    pmid = "17913474"
+    bad_src = join(test_weird_data_path, "17913474_recaptcha.html")
+    good_src = join(test_weird_data_path, "17913474_pond.html")
+
+    bad_dir = tmp_path / "a_bad"
+    good_dir = tmp_path / "b_good"
+    bad_dir.mkdir()
+    good_dir.mkdir()
+
+    # Same PMID filename in two different source folders
+    bad_target = bad_dir / f"{pmid}.html"
+    good_target = good_dir / f"{pmid}.html"
+    shutil.copy(bad_src, bad_target)
+    shutil.copy(good_src, good_target)
+
+    db_path = f"sqlite:///{(tmp_path / 'ace_dupe_pick_best.db').as_posix()}"
+    db = database.Database(adapter='sqlite', db_name=db_path)
+
+    # Intentionally place blocked/challenge page first.
+    ingest.add_articles(
+        db,
+        [str(bad_target), str(good_target)],
+        pmid_filenames=True,
+        force_ingest=True,
+        num_workers=1,
+        skip_metadata=True,
+    )
+
+    assert len(db.articles) == 1
+    assert len(db.articles[0].tables) >= 1
+    assert _count_valid_activations(db.articles[0].tables) >= 1
+
+
+def test_validate_scrape_flags_recaptcha_challenge_page(test_weird_data_path):
+    html = open(join(test_weird_data_path, "17913474_recaptcha.html")).read()
+    assert scrape._validate_scrape(html) is False
+
+
+@pytest.mark.parametrize(
+    "pmid,expected_source",
+    [
+        ("17088334", "PMCSource"),
+        ("26342221", "OUPSource"),
+        ("27623361", "ScienceDirectSource"),
+        ("27319001", "SpringerSource"),
+        ("20350171", "JournalOfCognitiveNeuroscienceSource"),
+        ("12860777", None),  # Unknown source -> DefaultSource fallback
+    ],
+)
+def test_additional_missed_in_main_text_regressions(test_weird_data_path, source_manager, pmid, expected_source):
+    html = open(join(test_weird_data_path, pmid + ".html")).read()
+    source = source_manager.identify_source(html)
+
+    if expected_source is None:
+        assert source is None
+        parser = source_manager.default_source
+        assert parser is not None
+    else:
+        assert source is not None
+        assert source.__class__.__name__ == expected_source
+        parser = source
+
+    article = parser.parse_article(html, pmid=pmid, skip_metadata=True)
+    assert article is not None
+    assert len(article.tables) >= 1
+    assert _count_valid_activations(article.tables) >= 1