Spaces:

PabloLE
/

Busqueda-Arxiv

Sleeping

Busqueda-Arxiv / src /detect_artifacts.py

Deploy: dashboard Streamlit + corpus 28924 papers + embeddings SPECTER

e3d7521 verified 5 months ago

7.14 kB

"""
Detector de codigo/datos basado en patrones regex sobre:
    - abstract (arXiv)
    - openalex_abstract (a veces la version del journal, mas pulida)
    - comment (campo "comments" de arXiv: ahi suelen ir links a github)
Estrategia:
  - URL patterns: hosts conocidos. Si aparece -> has_code/has_data = True
    + se guarda la URL deduplicada.
  - Promise patterns: frases que prometen disponibilidad ("code is available",
    "we release the dataset", etc.). Si aparece pero no hay URL -> flag True,
    URL list vacia. (Util porque a veces el link esta en el PDF, no en metadata.)
Limitaciones conocidas:
  - No parsea el PDF (eso queda para v2). Cobertura limitada al ~5-15% en q-fin.
  - Falsos positivos posibles: "github.com" mencionado en passing sin ser
    realmente el repo del paper. Costo aceptable para un MVP.
Uso:
    from src.detect_artifacts import detect, detect_for_db
    result = detect("Code: https://github.com/foo/bar. Dataset on Zenodo.")
    # -> {"has_code": True, "has_data": True, "code_urls": [...], "data_urls": [...]}
"""

from __future__ import annotations

import logging
import re
import sqlite3
from dataclasses import dataclass, field
from typing import Iterable

log = logging.getLogger(__name__)


# ------------------------------------------------------------
# Patrones de URL (hosts conocidos)
# ------------------------------------------------------------

# Permite paths con / . _ - y digitos. Corta en whitespace o paren/coma/comilla.
_URL_TAIL = r"[A-Za-z0-9_./\-]+"

CODE_URL_PATTERNS = [
    re.compile(rf"https?://(?:www\.)?github\.com/{_URL_TAIL}", re.IGNORECASE),
    re.compile(rf"https?://(?:www\.)?gitlab\.com/{_URL_TAIL}", re.IGNORECASE),
    re.compile(rf"https?://(?:www\.)?bitbucket\.org/{_URL_TAIL}", re.IGNORECASE),
    re.compile(rf"https?://(?:www\.)?codeocean\.com/{_URL_TAIL}", re.IGNORECASE),
    re.compile(rf"https?://(?:www\.)?huggingface\.co/{_URL_TAIL}", re.IGNORECASE),
    # github sin esquema explicito (muy comun en comments)
    re.compile(rf"(?<![\w/])github\.com/{_URL_TAIL}", re.IGNORECASE),
]

DATA_URL_PATTERNS = [
    re.compile(rf"https?://(?:www\.)?zenodo\.org/{_URL_TAIL}", re.IGNORECASE),
    re.compile(rf"https?://(?:www\.)?figshare\.com/{_URL_TAIL}", re.IGNORECASE),
    re.compile(rf"https?://(?:www\.)?osf\.io/{_URL_TAIL}", re.IGNORECASE),
    re.compile(rf"https?://dataverse\.[A-Za-z0-9.\-]+/{_URL_TAIL}", re.IGNORECASE),
    re.compile(rf"https?://(?:www\.)?kaggle\.com/datasets/{_URL_TAIL}", re.IGNORECASE),
    re.compile(rf"https?://(?:www\.)?data\.mendeley\.com/{_URL_TAIL}", re.IGNORECASE),
]

# Datasets en HuggingFace tienen URL distinta -> los tratamos como data
HF_DATASET_RE = re.compile(
    rf"https?://(?:www\.)?huggingface\.co/datasets/{_URL_TAIL}", re.IGNORECASE
)


# ------------------------------------------------------------
# Patrones de promesa (sin URL explicita)
# ------------------------------------------------------------

# Compilamos con \b para evitar matches dentro de palabras.
CODE_PROMISE_PATTERNS = [
    re.compile(r"\bcode\s+is\s+(publicly\s+|freely\s+)?available", re.IGNORECASE),
    re.compile(r"\b(source\s+)?code\s+(can\s+be|is|will\s+be)\s+(found|accessed|downloaded|made\s+available)", re.IGNORECASE),
    re.compile(r"\bwe\s+(make|release|provide|publish)\s+(the\s+|our\s+)?(source\s+)?code", re.IGNORECASE),
    re.compile(r"\breplication\s+(package|code|kit|materials)\b", re.IGNORECASE),
    re.compile(r"\bopen[\s-]?sourced?\b", re.IGNORECASE),
    re.compile(r"\bimplementation\s+is\s+(publicly\s+)?available", re.IGNORECASE),
]

DATA_PROMISE_PATTERNS = [
    re.compile(r"\b(data|dataset|datasets)\s+(is|are)\s+(publicly\s+|freely\s+)?available", re.IGNORECASE),
    re.compile(r"\bwe\s+(release|provide|publish|share)\s+(the\s+|our\s+|a\s+new\s+)?(data|dataset|datasets)", re.IGNORECASE),
    re.compile(r"\b(public|open)\s+dataset\b", re.IGNORECASE),
    re.compile(r"\bsupplementary\s+(materials?|data)\b", re.IGNORECASE),
]


# ------------------------------------------------------------
# Detection
# ------------------------------------------------------------

@dataclass
class Detection:
    has_code: bool = False
    has_data: bool = False
    code_urls: list[str] = field(default_factory=list)
    data_urls: list[str] = field(default_factory=list)


def _clean_url(u: str) -> str:
    """Quita puntuacion terminal tipica de prosa: . , ; ) etc."""
    return u.rstrip(".,;)>'\"`")


def detect(*texts: str | None) -> Detection:
    """Aplica todos los patrones sobre la union de los textos provistos."""
    blob = " \n ".join(t for t in texts if t)
    if not blob:
        return Detection()

    code_urls: set[str] = set()
    data_urls: set[str] = set()

    # HF datasets ANTES que HF general, para clasificarlo como data, no code.
    for m in HF_DATASET_RE.finditer(blob):
        data_urls.add(_clean_url(m.group(0)))

    for pat in CODE_URL_PATTERNS:
        for m in pat.finditer(blob):
            url = _clean_url(m.group(0))
            # Excluir HF datasets que ya cayeron en data
            if "/datasets/" in url.lower():
                continue
            code_urls.add(url)

    for pat in DATA_URL_PATTERNS:
        for m in pat.finditer(blob):
            data_urls.add(_clean_url(m.group(0)))

    has_code = bool(code_urls) or any(p.search(blob) for p in CODE_PROMISE_PATTERNS)
    has_data = bool(data_urls) or any(p.search(blob) for p in DATA_PROMISE_PATTERNS)

    return Detection(
        has_code=has_code,
        has_data=has_data,
        code_urls=sorted(code_urls),
        data_urls=sorted(data_urls),
    )


# ------------------------------------------------------------
# DB sweep
# ------------------------------------------------------------

@dataclass
class DetectStats:
    scanned: int = 0
    with_code: int = 0
    with_data: int = 0
    with_code_url: int = 0
    with_data_url: int = 0


def detect_for_db(
    conn: sqlite3.Connection,
    only_missing: bool = True,
) -> DetectStats:
    """
    Corre el detector sobre los papers de la DB.
    only_missing=True salta los papers que ya tienen has_code/has_data setteado
    (utiles para corridas incrementales).
    """
    from src import db

    where = "WHERE has_code IS NULL AND has_data IS NULL" if only_missing else ""
    rows = conn.execute(
        f"SELECT arxiv_id, abstract, openalex_abstract, comment "
        f"FROM papers {where}"
    ).fetchall()

    stats = DetectStats(scanned=len(rows))
    log.info("Escaneando %d papers...", len(rows))

    for r in rows:
        det = detect(r["abstract"], r["openalex_abstract"], r["comment"])
        db.update_artifacts(
            conn,
            arxiv_id=r["arxiv_id"],
            has_code=det.has_code,
            has_data=det.has_data,
            code_urls=det.code_urls,
            data_urls=det.data_urls,
        )
        if det.has_code:
            stats.with_code += 1
        if det.has_data:
            stats.with_data += 1
        if det.code_urls:
            stats.with_code_url += 1
        if det.data_urls:
            stats.with_data_url += 1

    conn.commit()
    return stats

	"""
	Detector de codigo/datos basado en patrones regex sobre:
	- abstract (arXiv)
	- openalex_abstract (a veces la version del journal, mas pulida)
	- comment (campo "comments" de arXiv: ahi suelen ir links a github)

	Estrategia:
	- URL patterns: hosts conocidos. Si aparece -> has_code/has_data = True
	+ se guarda la URL deduplicada.
	- Promise patterns: frases que prometen disponibilidad ("code is available",
	"we release the dataset", etc.). Si aparece pero no hay URL -> flag True,
	URL list vacia. (Util porque a veces el link esta en el PDF, no en metadata.)

	Limitaciones conocidas:
	- No parsea el PDF (eso queda para v2). Cobertura limitada al ~5-15% en q-fin.
	- Falsos positivos posibles: "github.com" mencionado en passing sin ser
	realmente el repo del paper. Costo aceptable para un MVP.

	Uso:
	from src.detect_artifacts import detect, detect_for_db
	result = detect("Code: https://github.com/foo/bar. Dataset on Zenodo.")
	# -> {"has_code": True, "has_data": True, "code_urls": [...], "data_urls": [...]}
	"""

	from __future__ import annotations

	import logging
	import re
	import sqlite3
	from dataclasses import dataclass, field
	from typing import Iterable

	log = logging.getLogger(__name__)


	# ------------------------------------------------------------
	# Patrones de URL (hosts conocidos)
	# ------------------------------------------------------------

	# Permite paths con / . _ - y digitos. Corta en whitespace o paren/coma/comilla.
	_URL_TAIL = r"[A-Za-z0-9_./\-]+"

	CODE_URL_PATTERNS = [
	re.compile(rf"https?://(?:www\.)?github\.com/{_URL_TAIL}", re.IGNORECASE),
	re.compile(rf"https?://(?:www\.)?gitlab\.com/{_URL_TAIL}", re.IGNORECASE),
	re.compile(rf"https?://(?:www\.)?bitbucket\.org/{_URL_TAIL}", re.IGNORECASE),
	re.compile(rf"https?://(?:www\.)?codeocean\.com/{_URL_TAIL}", re.IGNORECASE),
	re.compile(rf"https?://(?:www\.)?huggingface\.co/{_URL_TAIL}", re.IGNORECASE),
	# github sin esquema explicito (muy comun en comments)
	re.compile(rf"(?<![\w/])github\.com/{_URL_TAIL}", re.IGNORECASE),
	]

	DATA_URL_PATTERNS = [
	re.compile(rf"https?://(?:www\.)?zenodo\.org/{_URL_TAIL}", re.IGNORECASE),
	re.compile(rf"https?://(?:www\.)?figshare\.com/{_URL_TAIL}", re.IGNORECASE),
	re.compile(rf"https?://(?:www\.)?osf\.io/{_URL_TAIL}", re.IGNORECASE),
	re.compile(rf"https?://dataverse\.[A-Za-z0-9.\-]+/{_URL_TAIL}", re.IGNORECASE),
	re.compile(rf"https?://(?:www\.)?kaggle\.com/datasets/{_URL_TAIL}", re.IGNORECASE),
	re.compile(rf"https?://(?:www\.)?data\.mendeley\.com/{_URL_TAIL}", re.IGNORECASE),
	]

	# Datasets en HuggingFace tienen URL distinta -> los tratamos como data
	HF_DATASET_RE = re.compile(
	rf"https?://(?:www\.)?huggingface\.co/datasets/{_URL_TAIL}", re.IGNORECASE
	)


	# ------------------------------------------------------------
	# Patrones de promesa (sin URL explicita)
	# ------------------------------------------------------------

	# Compilamos con \b para evitar matches dentro de palabras.
	CODE_PROMISE_PATTERNS = [
	re.compile(r"\bcode\s+is\s+(publicly\s+\|freely\s+)?available", re.IGNORECASE),
	re.compile(r"\b(source\s+)?code\s+(can\s+be\|is\|will\s+be)\s+(found\|accessed\|downloaded\|made\s+available)", re.IGNORECASE),
	re.compile(r"\bwe\s+(make\|release\|provide\|publish)\s+(the\s+\|our\s+)?(source\s+)?code", re.IGNORECASE),
	re.compile(r"\breplication\s+(package\|code\|kit\|materials)\b", re.IGNORECASE),
	re.compile(r"\bopen[\s-]?sourced?\b", re.IGNORECASE),
	re.compile(r"\bimplementation\s+is\s+(publicly\s+)?available", re.IGNORECASE),
	]

	DATA_PROMISE_PATTERNS = [
	re.compile(r"\b(data\|dataset\|datasets)\s+(is\|are)\s+(publicly\s+\|freely\s+)?available", re.IGNORECASE),
	re.compile(r"\bwe\s+(release\|provide\|publish\|share)\s+(the\s+\|our\s+\|a\s+new\s+)?(data\|dataset\|datasets)", re.IGNORECASE),
	re.compile(r"\b(public\|open)\s+dataset\b", re.IGNORECASE),
	re.compile(r"\bsupplementary\s+(materials?\|data)\b", re.IGNORECASE),
	]


	# ------------------------------------------------------------
	# Detection
	# ------------------------------------------------------------

	@dataclass
	class Detection:
	has_code: bool = False
	has_data: bool = False
	code_urls: list[str] = field(default_factory=list)
	data_urls: list[str] = field(default_factory=list)


	def _clean_url(u: str) -> str:
	"""Quita puntuacion terminal tipica de prosa: . , ; ) etc."""
	return u.rstrip(".,;)>'\"`")


	def detect(*texts: str \| None) -> Detection:
	"""Aplica todos los patrones sobre la union de los textos provistos."""
	blob = " \n ".join(t for t in texts if t)
	if not blob:
	return Detection()

	code_urls: set[str] = set()
	data_urls: set[str] = set()

	# HF datasets ANTES que HF general, para clasificarlo como data, no code.
	for m in HF_DATASET_RE.finditer(blob):
	data_urls.add(_clean_url(m.group(0)))

	for pat in CODE_URL_PATTERNS:
	for m in pat.finditer(blob):
	url = _clean_url(m.group(0))
	# Excluir HF datasets que ya cayeron en data
	if "/datasets/" in url.lower():
	continue
	code_urls.add(url)

	for pat in DATA_URL_PATTERNS:
	for m in pat.finditer(blob):
	data_urls.add(_clean_url(m.group(0)))

	has_code = bool(code_urls) or any(p.search(blob) for p in CODE_PROMISE_PATTERNS)
	has_data = bool(data_urls) or any(p.search(blob) for p in DATA_PROMISE_PATTERNS)

	return Detection(
	has_code=has_code,
	has_data=has_data,
	code_urls=sorted(code_urls),
	data_urls=sorted(data_urls),
	)


	# ------------------------------------------------------------
	# DB sweep
	# ------------------------------------------------------------

	@dataclass
	class DetectStats:
	scanned: int = 0
	with_code: int = 0
	with_data: int = 0
	with_code_url: int = 0
	with_data_url: int = 0


	def detect_for_db(
	conn: sqlite3.Connection,
	only_missing: bool = True,
	) -> DetectStats:
	"""
	Corre el detector sobre los papers de la DB.

	only_missing=True salta los papers que ya tienen has_code/has_data setteado
	(utiles para corridas incrementales).
	"""
	from src import db

	where = "WHERE has_code IS NULL AND has_data IS NULL" if only_missing else ""
	rows = conn.execute(
	f"SELECT arxiv_id, abstract, openalex_abstract, comment "
	f"FROM papers {where}"
	).fetchall()

	stats = DetectStats(scanned=len(rows))
	log.info("Escaneando %d papers...", len(rows))

	for r in rows:
	det = detect(r["abstract"], r["openalex_abstract"], r["comment"])
	db.update_artifacts(
	conn,
	arxiv_id=r["arxiv_id"],
	has_code=det.has_code,
	has_data=det.has_data,
	code_urls=det.code_urls,
	data_urls=det.data_urls,
	)
	if det.has_code:
	stats.with_code += 1
	if det.has_data:
	stats.with_data += 1
	if det.code_urls:
	stats.with_code_url += 1
	if det.data_urls:
	stats.with_data_url += 1

	conn.commit()
	return stats