Spaces:
Running
Running
File size: 5,918 Bytes
aa0a3e7 7a1bb2f aa0a3e7 7a1bb2f aa0a3e7 | 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 | import calendar
import hashlib
import json
import re
from typing import Any
from pydantic import BaseModel, ConfigDict, Field
class ParseOCR(BaseModel):
model_config = ConfigDict(extra="forbid")
content_thresholds: list[float] | None = Field(default=None, min_length=4, max_length=4)
auto_vsplit: bool | None = None
vsplit: float | None = Field(default=None, ge=0, le=1)
class ParseDate(BaseModel):
model_config = ConfigDict(extra="forbid")
year: int | None = Field(default=None, ge=1, le=9999)
month: int | None = Field(default=None, ge=1, le=12)
day: int | None = Field(default=None, ge=1, le=31)
class ParseArticle(BaseModel):
model_config = ConfigDict(extra="forbid")
title: str = Field(min_length=1, max_length=500)
authors: list[str] = Field(default_factory=list, max_length=50)
dates: list[ParseDate] = Field(default_factory=list, max_length=50)
is_range_date: bool = False
page_start: int = Field(ge=1, le=10000)
page_end: int = Field(ge=1, le=10000)
content: str = Field(min_length=1, max_length=200000)
base_part_count: int = Field(ge=1, le=15000)
ocr: ParseOCR | None = None
ocr_exceptions: dict[str, ParseOCR] = Field(default_factory=dict, max_length=100)
class ParseRequest(BaseModel):
model_config = ConfigDict(extra="forbid")
doc_id: str = Field(min_length=1, max_length=512)
base_content_hash: str | None = Field(default=None, min_length=64, max_length=64, pattern=r"[0-9a-f]{64}")
articles: list[ParseArticle] = Field(min_length=1, max_length=200)
description: str = Field(default="", max_length=2000)
def queue_item(source: dict[str, Any]) -> dict[str, Any]:
article = source.get("article") or {}
return {
"doc_id": source.get("doc_id"),
"archive_id": source.get("archive_id"),
"publication_id": source.get("publication_id"),
"publication_name": source.get("publication_name"),
"publication_type": source.get("publication_type"),
"title": source.get("title"),
"authors": source.get("authors") or [],
"date_display": source.get("date_display") or [],
"page_start": article.get("page_start"),
"page_end": article.get("page_end"),
"source_files": source.get("source_files") or [],
}
def is_queue_source(source: dict[str, Any] | None, archive_id: int, prefix: str) -> bool:
return bool(
source
and source.get("archive_id") == archive_id
and str(source.get("title") or "").startswith(prefix)
)
def validate_articles(
articles: list[ParseArticle],
source: dict[str, Any],
queue_prefix: str,
) -> tuple[list[dict[str, Any]] | None, str | None]:
current = source.get("article") or {}
source_start = int(current.get("page_start") or 0)
source_end = int(current.get("page_end") or 0)
normalized = []
covered = set()
article_ids = set()
for article in articles:
if article.title.strip().startswith(queue_prefix):
return None, "article title must replace the placeholder"
if not article.content.strip():
return None, "article content cannot be empty"
if article.page_start > article.page_end or article.page_start < source_start or article.page_end > source_end:
return None, "article page range is outside the source range"
if any(not author.strip() or len(author.strip()) > 300 for author in article.authors):
return None, "article author is invalid"
for page in range(article.page_start, article.page_end + 1):
covered.add(page)
item = article.model_dump(exclude_none=True)
item["title"] = item["title"].strip()
item["authors"] = [author.strip() for author in item.get("authors", [])]
item["dates"] = [date for date in item.get("dates", []) if date]
for date in item["dates"]:
if date.get("day") is not None and date.get("month") is None:
return None, "article day requires a month"
if date.get("month") is not None and date.get("year") is None:
return None, "article month requires a year"
if date.get("day") is not None and date["day"] > calendar.monthrange(date["year"], date["month"])[1]:
return None, "article date is not a real calendar date"
ocr_values = [item.get("ocr") or {}, *(item.get("ocr_exceptions") or {}).values()]
for options in ocr_values:
thresholds = options.get("content_thresholds")
if thresholds is not None and (
any(value < 0 or value > 1 for value in thresholds)
or thresholds[0] + thresholds[1] >= 1
or thresholds[2] + thresholds[3] >= 1
):
return None, "article OCR crop is invalid"
for raw_page in item.get("ocr_exceptions") or {}:
if not re.fullmatch(r"[1-9][0-9]*", raw_page) or not article.page_start <= int(raw_page) <= article.page_end:
return None, "article OCR exception page is invalid"
dates = sorted(
f"{date.get('year') or '0000'}-{int(date.get('month') or 0):02d}-{int(date.get('day') or 0):02d}"
for date in item["dates"]
)
identity = hashlib.md5(json.dumps([
item["title"], dates, bool(item.get("is_range_date")), sorted(item["authors"]), "",
], ensure_ascii=False, separators=(",", ":")).encode()).hexdigest()[:10]
if identity in article_ids:
return None, "articles would generate duplicate article IDs"
article_ids.add(identity)
normalized.append(item)
expected = set(range(source_start, source_end + 1))
if covered != expected:
missing = sorted(expected - covered)
return None, f"article ranges do not cover pages: {', '.join(map(str, missing))}"
return normalized, None
|