Spaces:
Running on Zero
Running on Zero
File size: 6,782 Bytes
413d5a1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 | """Deterministic loading and validation for source article records."""
from __future__ import annotations
import json
from collections import Counter
from pathlib import Path
from typing import Any
from pydantic import ValidationError
from gcmd_classifier.errors import (
ArticleFileNotFoundError,
ArticleJSONDecodeError,
ArticleTopLevelTypeError,
)
from gcmd_classifier.models import ArticleLoadResult, ArticleRecord, ArticleValidationIssue
REQUIRED_FIELDS = ("DOI", "Title", "Year", "Abstract")
_STRING_FIELDS = ("DOI", "Title", "Abstract")
_NON_EMPTY_STRING_FIELDS = ("DOI", "Title")
def read_article_json(path: str | Path) -> Any:
"""Read and decode an article JSON file without validating record contents."""
article_path = Path(path)
try:
raw_bytes = article_path.read_bytes()
except FileNotFoundError as exc:
raise ArticleFileNotFoundError(f"Article file not found: {article_path}") from exc
try:
return json.loads(raw_bytes)
except json.JSONDecodeError as exc:
raise ArticleJSONDecodeError(f"Invalid JSON in article file: {article_path}") from exc
def validate_article_record(
raw_record: Any, index: int
) -> tuple[ArticleRecord | None, tuple[ArticleValidationIssue, ...]]:
"""Validate one article record, preserving source values exactly on success."""
issues: list[ArticleValidationIssue] = []
if not isinstance(raw_record, dict):
return None, (
ArticleValidationIssue(
index=index,
code="ARTICLE_NOT_OBJECT",
message="Article entry must be a JSON object.",
),
)
for field in REQUIRED_FIELDS:
if field not in raw_record:
issues.append(
ArticleValidationIssue(
index=index,
code="MISSING_REQUIRED_FIELD",
field=field,
message=f"Article record is missing required field {field!r}.",
DOI=_safe_doi(raw_record),
)
)
for field in _STRING_FIELDS:
if field in raw_record and not isinstance(raw_record[field], str):
issues.append(
ArticleValidationIssue(
index=index,
code="INVALID_FIELD_TYPE",
field=field,
message=f"Article field {field!r} must be a string.",
DOI=_safe_doi(raw_record),
)
)
for field in _NON_EMPTY_STRING_FIELDS:
if field in raw_record and isinstance(raw_record[field], str) and not raw_record[field]:
issues.append(
ArticleValidationIssue(
index=index,
code="EMPTY_REQUIRED_STRING",
field=field,
message=f"Article field {field!r} must be a non-empty string.",
DOI=_safe_doi(raw_record),
)
)
if "Year" in raw_record:
year = raw_record["Year"]
if isinstance(year, bool) or not isinstance(year, int):
issues.append(
ArticleValidationIssue(
index=index,
code="INVALID_FIELD_TYPE",
field="Year",
message="Article field 'Year' must be a strict integer and not a Boolean.",
DOI=_safe_doi(raw_record),
)
)
unexpected_fields = tuple(field for field in raw_record if field not in REQUIRED_FIELDS)
for field in unexpected_fields:
issues.append(
ArticleValidationIssue(
index=index,
code="UNEXPECTED_FIELD",
field=field,
message=f"Unexpected article field {field!r} is not allowed in the MVP.",
DOI=_safe_doi(raw_record),
)
)
if issues:
return None, tuple(issues)
try:
return ArticleRecord.model_validate(raw_record), ()
except ValidationError as exc:
return None, tuple(_issues_from_pydantic_error(exc, raw_record=raw_record, index=index))
def validate_article_records(raw_records: Any) -> ArticleLoadResult:
"""Validate a top-level article list and collect all record-level validation errors."""
if not isinstance(raw_records, list):
raise ArticleTopLevelTypeError("Article source JSON must be a top-level list.")
valid_candidates: list[tuple[int, ArticleRecord]] = []
errors: list[ArticleValidationIssue] = []
for index, raw_record in enumerate(raw_records):
article, issues = validate_article_record(raw_record, index)
errors.extend(issues)
if article is not None:
valid_candidates.append((index, article))
doi_counts = Counter(article.DOI for _, article in valid_candidates)
duplicate_dois = {doi for doi, count in doi_counts.items() if count > 1}
if duplicate_dois:
unique_candidates: list[tuple[int, ArticleRecord]] = []
for index, article in valid_candidates:
if article.DOI in duplicate_dois:
errors.append(
ArticleValidationIssue(
index=index,
code="DUPLICATE_DOI",
field="DOI",
message=f"Duplicate DOI value {article.DOI!r}.",
DOI=article.DOI,
)
)
else:
unique_candidates.append((index, article))
valid_candidates = unique_candidates
return ArticleLoadResult(
articles=tuple(article for _, article in valid_candidates),
errors=tuple(errors),
source_count=len(raw_records),
)
def load_articles(path: str | Path) -> ArticleLoadResult:
"""Read and validate article records from a JSON file."""
return validate_article_records(read_article_json(path))
def _safe_doi(raw_record: dict[str, Any]) -> str | None:
value = raw_record.get("DOI")
return value if isinstance(value, str) else None
def _issues_from_pydantic_error(
exc: ValidationError, *, raw_record: dict[str, Any], index: int
) -> tuple[ArticleValidationIssue, ...]:
issues: list[ArticleValidationIssue] = []
for error in exc.errors():
location = error.get("loc", ())
field = str(location[0]) if location else None
issues.append(
ArticleValidationIssue(
index=index,
code="MODEL_VALIDATION_ERROR",
field=field,
message=str(error.get("msg", "Article model validation failed.")),
DOI=_safe_doi(raw_record),
)
)
return tuple(issues)
|