christian/smk-playground
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Complete Milestone 1 audited source ingest

Browse source
This commit is contained in:
Christian Johannesen 2026-10-08 00:24:51 +02:00
parent c401957b51
commit 1114ccd8a7
17 changed files with 1040 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
AGENTS.md
View file

@ -164,6 +164,9 @@ Parquet is the portable source of truth for derived tabular output. DuckDB may p
- Do not run duplicate visual feature extraction just because both language snapshots contain the same artwork. - Do not run duplicate visual feature extraction just because both language snapshots contain the same artwork.
- If language-independent values disagree, preserve provenance and report/resolve via an explicit deterministic rule. - If language-independent values disagree, preserve provenance and report/resolve via an explicit deterministic rule.
- Missing one language must not automatically discard an otherwise valid artwork. - Missing one language must not automatically discard an otherwise valid artwork.
- Use `smk:<language>:<id>` as source-row identity for the audited snapshots. Never use array
position. Treat duplicate source-row IDs as errors and repeated `object_number` values as
unresolved groups until an explicit canonical rule handles them.
## Canonical artwork rules ## Canonical artwork rules

19
README.md
View file

@ -121,4 +121,21 @@ See:
## Current status ## Current status
Repository foundation and Dataset v1 specification. The next action is source reconnaissance against the two real local JSON files before implementing the final parser. Milestones 0–1 complete: the real source structure is manifested, both snapshots have
bounded-memory source Parquets, and duplicate identity, dimensions, palette/visual quality,
rights and DA/EN disagreements are audited. Canonical artwork generation is next; visual feature
extraction has not started.
Audit results are documented in:
- [`docs/DUPLICATE_AUDIT.md`](docs/DUPLICATE_AUDIT.md)
- [`docs/DIMENSIONS.md`](docs/DIMENSIONS.md)
- [`docs/SOURCE_DISAGREEMENTS.md`](docs/SOURCE_DISAGREEMENTS.md)
- [`docs/PALETTE_VISUAL_AUDIT.md`](docs/PALETTE_VISUAL_AUDIT.md)
- [`docs/RIGHTS_AUDIT.md`](docs/RIGHTS_AUDIT.md)
Run the audited ingest with:
```bash
smk milestone1
```

26
docs/DATA_MODEL.md
View file

@ -60,10 +60,34 @@ Corpus-wide counts, missingness, exclusions, warnings and distribution diagnosti
## Identity ## Identity
Prefer `object_number` as the stable artwork key where present. Detect and report missing/duplicate IDs before silently deduplicating. `object_number` is an artwork-facing identity but is not a source-row key: 305 DA and 306 EN
groups contain more than one record. SMK's top-level `id` is present and unique for every row in
each audited snapshot (202,728 DA; 203,023 EN).
Source rows therefore use:
```text
source_row_id = "smk:" + source_language + ":" + id
```
The language prefix distinguishes the DA and EN representations of the same SMK source record.
If `id` is absent in a future source, use a deterministic SHA-256 of language + stable source
content and report the fallback. Array index is never permanent identity. A duplicate
`source_row_id` is a hard ingest error.
Keep `smk_source_id`, `object_number`, language, source-file hash and logical record hash alongside
the internal ID. Canonical artwork identity is deliberately deferred until duplicate groups have
an explicit resolution; it must not silently equate every repeated `object_number` row.
If duplicate IDs exist, inspect them and implement an explicit deterministic policy with provenance; do not keep an arbitrary “first row”. If duplicate IDs exist, inspect them and implement an explicit deterministic policy with provenance; do not keep an arbitrary “first row”.
## Manifest-bound source Parquet
`source_da.parquet` and `source_en.parquet` contain one row per source row, including selected typed
columns, audit JSON columns and `source_record_json`, which preserves the complete parsed record.
This is logical JSON preservation, not original byte-layout preservation. Parquet metadata records
the source-file and source-manifest SHA-256 values.
## Null semantics ## Null semantics
Null means unavailable/not safely derivable. Null means unavailable/not safely derivable.

74
docs/DIMENSIONS.md Normal file
View file

@ -0,0 +1,74 @@
# Dimension vocabulary audit
The complete DA and EN snapshots were profiled before implementing canonical geometry. The detailed
frequency/cross-tab dataset is `data/reports/dimension-vocabulary.parquet`.
## Coverage and quality
| Measure | DA | EN |
|---|---:|---:|
| Parsed dimension entries | 430,613 | 431,230 |
| Numeric text values | 430,429 | 431,044 |
| Missing values | 184 | 186 |
| Records without a dimension array | 3,587 | 3,581 |
| Missing `part` entries | 2,715 | 2,732 |
| Missing `type` entries | 265 | 265 |
| Missing `unit` entries | 292 | 310 |
| Exact duplicate measurements beyond first | 23 | 24 |
| Conflicting values for same row/part/type/unit | 413 | 413 |
No non-empty malformed numeric text was observed: every present `value` matched the strict numeric
grammar. Missing values remain missing. Conflicts and duplicate measurements still require
row-level diagnostics during canonicalization.
## Vocabulary
The language files localize type/unit more consistently than part:
| Canonical concept | DA observed | EN observed |
|---|---|---|
| width | `bredde` (208,330) | `width` (208,636) |
| height | `højde` (209,736) | `height` (210,033) |
| depth | `dybde` (11,301) | `depth` (11,306) |
| diameter | `diameter` (913) | `diameter` (914) |
| length | `længde` (22) | `length` (31) |
| millimetres | `millimeter` (396,316), `mm` (22), typo `milimeter` (520) | `mm` (397,447) |
| centimetres | `centimeter` (33,402), `cm` (2) | `cm` (33,434) |
Non-geometric types include weight (`vægt`/`weight`, 39 each), playing time
(`spilletid`/`playing time`, 5 each), and one `bruttoregisterton`. Their units include kg/kilo,
minutes and seconds; they must not enter width × height calculations.
The largest part labels are sheet (`bladmaal`, `bladmål`, `Bladmål`), plate
(`plademaal`, `plademål`, `Plademål`), `Netto`/`netto`, mounting, image and `Brutto`/`brutto`.
Both language snapshots retain Danish part labels. Case, `aa`/`å`, and `*mål` spelling variants
are source variation rather than different physical concepts.
## Proposed explicit mapping
Canonicalization should normalize Unicode, trim, case-fold and map only listed aliases:
| Source aliases | Canonical part/axis/unit |
|---|---|
| `Netto`, `netto`, `Nettomål` | `net` part |
| `Brutto`, `brutto`, `Bruttomål`, `bruttomål` | `gross` part |
| `bladmaal`, `bladmål`, `Bladmål`, `Bladmaal` | `sheet` part |
| `plademaal`, `plademål`, `Plademål`, `Plademaal` | `plate` part |
| `monteringsmaal`, `monteringsmål` | `mount` part |
| `billedmaal`, `billedmål`, `Billedmål` | `image` part |
| `papirmål` | `paper` part |
| `bredde`, `width` | `width` |
| `højde`, `height` | `height` |
| `dybde`, `depth` | `depth` |
| `diameter` | `diameter` |
| `længde`, `length` | `length` |
| `millimeter`, `milimeter`, `mm` | `mm`, factor to cm `0.1` |
| `centimeter`, `cm` | `cm`, factor to cm `1.0` |
Prefer a complete, non-conflicting `net` width/height pair. Keep plate, sheet, image, paper and
mount dimensions separately rather than silently treating them as equivalent. Gross is a fallback
only under an explicitly named policy. Diameter is not width; depth/diameter and object classes such
as sculpture, relief, installation, statue, object, audio/video and multi-part work make generic
width × height potentially misleading. Unknown parts/types/units stay raw and canonical geometry
stays null.

41
docs/DUPLICATE_AUDIT.md Normal file
View file

@ -0,0 +1,41 @@
# Duplicate `object_number` audit
Milestone 1 audited every duplicate group without dropping any source row. The detailed report is
`data/reports/duplicate-object-numbers.parquet` (1,280 occurrence rows, 19 columns).
## Findings
| | DA | EN |
|---|---:|---:|
| Duplicate `object_number` groups | 305 | 306 |
| Rows in duplicate groups | 639 | 641 |
| Groups of 2 rows | 297 | 298 |
| Groups of 3 rows | 7 | 7 |
| Groups of 24 rows | 1 | 1 |
No group is logically identical. Because every group differs logically, its source JSON bytes also
cannot be identical. Every occurrence has a distinct SMK `id`; these are not accidental repeated
array rows. The most common differing fields are `id` (all 1,280 rows), `created`, `titles`,
`dimensions`, `modified`, `documentation`, `inscriptions`, `production`, and production dates.
The audit classifies 605 language-specific groups as multiple SMK source records sharing an
`object_number`. Six groups have a `verso` identity and are marked as possible alternate/subrecord
cases, but the label is diagnostic rather than a merge decision. The largest group is
`KKS1974-33/1`, with 24 distinct source records in each language. Seven groups per language have
three records; all other groups have two.
Examples demonstrate that the repeated number can cover substantially different objects:
- `DEP1064/93`: distinct `id`, creation time, inscriptions, techniques and titles;
- `DEP2387`: also differs in public-domain status, rights, dimensions, object names, production and
acquisition metadata;
- `KKS1974-33/1`: 24 records with distinct IDs, titles, dimensions and URLs.
## Handling rule
Do not deduplicate the source layer. Preserve every row under `source_row_id`. Before creating one
canonical artwork row, classify each repeated `object_number` group using IDs, titles, part/verso
relationships, related-object fields and URLs. Only collapse a group when an explicit rule proves
that the records are versions of one artwork. Otherwise assign a later canonical group identifier
or retain multiple canonical objects. A newest-`modified` or first-row rule is not safe.

5
docs/IMPLEMENTATION_PLAN.md
View file

@ -26,6 +26,11 @@ Done when we understand the actual dumps well enough to implement the parser fro
Done when a representative fixture/sample becomes deterministic canonical rows. Done when a representative fixture/sample becomes deterministic canonical rows.
Status: source loading and pre-canonical audits complete. Stable source-row identity, lossless
logical Parquet, duplicate taxonomy, dimension vocabulary, palette/visual quality and DA/EN
precedence recommendations are documented. Canonical artwork generation remains blocked on an
explicit duplicate-group resolution policy and is intentionally not performed in this milestone.
## Milestone 2 — Weighted palette + colour conversion ## Milestone 2 — Weighted palette + colour conversion
- locate ordinary/weighted palette fields in the actual snapshot/enrichment data - locate ordinary/weighted palette fields in the actual snapshot/enrichment data

40
docs/PALETTE_VISUAL_AUDIT.md Normal file
View file

@ -0,0 +1,40 @@
# Palette and SMK visual-field audit
The machine-readable profile is `data/reports/palette-visual-profile.json`.
## `colors`
`colors` is missing or not an array in 150,479/202,728 DA records (74.23%) and
150,774/203,023 EN records (74.26%). The 52,249 arrays in each language contain 1–19 colours;
the most common lengths are five (9,463), four (9,035), three (7,447), six (7,170) and two
(6,667). All values match strict six-digit `#RRGGBB` syntax.
One record, `KKS8001`, repeats colours within its palette:
`#F4C6A3` twice and `#9B9B9B` three times. Repetition must be preserved; without source
documentation it must not be silently interpreted as weight. Comparable DA/EN singleton palettes
have zero value or ordering disagreements.
## Visual fields
Counts and distribution summaries are identical across languages for the shared visual data:
| Field | Numeric | Median | P01 | P99 | Min | Max |
|---|---:|---:|---:|---:|---:|---:|
| brightness | 52,171 | 7.296492 | 2.311502 | 8.939394 | 0.575758 | 9.090909 |
| contrast | 52,171 | 3.461959 | 0.544415 | 7.249865 | 0.095043 | 9.672021 |
| saturation | 45,024 | 1.494608 | 0.141818 | 5.924530 | 0.000001 | 9.765914 |
| entropy | 52,249 | 9.513508 | 3.768011 | 9.792444 | 0.685058 | 9.792481 |
| colortemp | 52,237 | 3.411809 | 2.840766 | 5.679331 | -1.276565 | 2.674e27 |
The 12 textual `colortemp` values are all exactly `"Infinity"`, in both languages, affecting:
`KMS7655/12`, `KMS8868`, `KKS1995-40a/2`, `KKS1995-40a/1`, `KKS1995-40a/4`,
`KKSbog7658/4`, `KMS7739/3`, `KMS7820`, `KMS7830`, `KMS7768`, `KMS7901`, and
`KKSbog7658`.
They are malformed numeric values and remain explicit diagnostics. Numeric `colortemp` also has
serious outliers: 89 values are negative, 77 exceed 10, 19 exceed 100 and 11 exceed one million.
The largest is `KMS7561/4` at `2.6739636e27`. These values are preserved but must not be treated as
valid feature inputs without a documented validity range/source correction. No other visual field
has malformed values in the audit.

6
docs/PIPELINE.md
View file

@ -57,6 +57,12 @@ Before writing the final record parser:
- report duplicates/missing IDs, - report duplicates/missing IDs,
- create `source-manifest.json`. - create `source-manifest.json`.
## Stage 1 — audited source ingest
Milestone 1 streams `$[]` with `ijson`, rejects source-ID collisions, preserves every source row in
manifest-bound Parquet and produces duplicate, dimension and language-disagreement audits. It does
not yet collapse duplicate `object_number` groups or produce canonical artwork rows.
## Ingest performance rule ## Ingest performance rule
The source files are large. Use bounded-memory streaming where practical. A one-time conversion to Parquet/Arrow as an interim representation is encouraged if it simplifies all later stages. The source files are large. Use bounded-memory streaming where practical. A one-time conversion to Parquet/Arrow as an interim representation is encouraged if it simplifies all later stages.

38
docs/RIGHTS_AUDIT.md Normal file
View file

@ -0,0 +1,38 @@
# Rights and public-domain reconnaissance
The machine-readable profile is `data/reports/rights-public-domain-profile.json`. This audit does
not freeze `smk-public-visual-v1`.
## Rights values
| Value | DA | EN |
|---|---:|---:|
| Public Domain Mark 1.0 | 151,618 | 151,913 |
| CC0 1.0 | 598 | 598 |
| SMK use-material policy, localized URL | 50,512 | 50,512 |
The DA/EN rights disagreement is a URL-localization effect for comparable singleton records: DA
uses the Danish use-material policy URL and EN uses its English counterpart. `public_domain`
itself agrees across comparable records. Preserve the original URL but map these two URLs to one
policy concept for analysis.
## Availability combinations
| Snapshot | Public domain + image + public URL | Public domain, no image/URL | Not public domain + image + URL |
|---|---:|---:|---:|
| DA | 39,505 | 112,711 | 14,918 |
| EN | 39,506 | 113,005 | 14,920 |
There are also two non-public-domain rows per language with a public image URL while `has_image` is
false. Image and rights indicators should therefore be tested explicitly rather than inferred from
URL presence alone.
## Recommendation for the next milestone
A candidate “public visual artwork” predicate should require an explicitly public-domain rights
state (Public Domain Mark or CC0 plus `public_domain == true`), a coherent usable image bundle,
and a resolved canonical artwork identity. It should separately report object classification and
exclude/flag non-artwork or unsuitable visual classes through an explicit vocabulary, not an
assumption that every source record is a 2D artwork. The final rule and membership count remain
deliberately unfrozen until duplicate resolution and canonical artwork generation are complete.

13
docs/SOURCE_DATA.md
View file

@ -87,3 +87,16 @@ The public API and enrichment API are secondary sources in v1. Use them for:
Do not make tens of thousands of live requests merely to recreate data already present in the snapshots. Do not make tens of thousands of live requests merely to recreate data already present in the snapshots.
SMK states that its full JSON dataset download is updated nightly. Therefore every local dump is a time-specific snapshot; reproducibility depends on hashing and manifesting the exact files used. SMK states that its full JSON dataset download is updated nightly. Therefore every local dump is a time-specific snapshot; reproducibility depends on hashing and manifesting the exact files used.
## Milestone 1 source artifacts
The bounded-memory Python ingest uses `ijson.items(stream, "item")` and writes:
- `data/interim/source_da.parquet` — 202,728 rows;
- `data/interim/source_en.parquet` — 203,023 rows.
Every row preserves the complete parsed source record plus stable row identity and manifest/file
hash provenance. Generated audit Parquets cover duplicate object numbers, dimension vocabulary and
DA/EN source disagreements. See `DUPLICATE_AUDIT.md`, `DIMENSIONS.md` and
`SOURCE_DISAGREEMENTS.md`. Palette/visual quality and rights are summarized in
`PALETTE_VISUAL_AUDIT.md` and `RIGHTS_AUDIT.md`.

45
docs/SOURCE_DISAGREEMENTS.md Normal file
View file

@ -0,0 +1,45 @@
# DA/EN source disagreement audit
The detailed field-level report is `data/reports/source-disagreements.parquet`. It compares the
202,393 shared `object_number` identities while isolating 305 duplicate identity groups that first
need duplicate resolution.
## Findings
For comparable singleton records:
- `rights`: 50,425 differences, all the same localization pair: DA
`https://www.smk.dk/section/brug-af-museets-materiale/` versus EN
`https://www.smk.dk/en/section/use-of-smk-material/`;
- normalized dimensions: 156 differences;
- `has_image`: 7 differences;
- image width and height: 15 differences each (8 unequal, 5 missing in DA, 2 missing in EN);
- IIIF image IDs/info: 11 differences each (3 unequal, 6 missing in DA, 2 missing in EN);
- image thumbnail/native URLs: 15 differences each.
There were no comparable-singleton disagreements in `public_domain`, `colors`, brightness,
contrast, saturation, entropy or colortemp. Palette ordering therefore also agrees across the
language snapshots wherever both contain a comparable singleton record.
The image differences are not merely localization. Examples include `KKS1984-179` (2048×3072 DA
versus 5078×6618 EN with different IIIF IDs) and `KKSgb5300` (8180×5262 DA versus 999×643 EN).
The seven `has_image` disagreements go in both directions. Dimension differences include both
small numeric conflicts (`DEP977`) and measurements present in only one language.
## Source precedence recommendation
- Identity: match cross-language rows by SMK `id` where available and verify `object_number`; do
not use array position.
- Localized text: retain DA and EN independently with their explicit source language.
- `rights`: preserve both URLs, but map this exact localized pair to one policy concept. Use
`public_domain` plus the original rights URL for legal provenance.
- Palette and visual values: accept the shared value when equal. If a future snapshot disagrees,
flag it rather than apply global language precedence.
- Image metadata: choose a coherent image bundle (IIIF ID, width, height and URLs) at row level.
Prefer the bundle that is internally complete and consistent; if both differ and are complete,
retain both and require freshness/content verification. Never mix dimensions from one bundle
with an image ID from the other.
- Dimensions: compare localized canonical tuples, preserve both originals, and leave conflicting
canonical values unresolved unless an explicit part-aware rule decides them.
- Missing one language: retain the available source value with provenance.

1
pyproject.toml
View file

@ -28,6 +28,7 @@ smk = "smk_playground.cli:app"
[tool.pytest.ini_options] [tool.pytest.ini_options]
testpaths = ["tests"] testpaths = ["tests"]
pythonpath = ["src"]
[tool.ruff] [tool.ruff]
line-length = 100 line-length = 100

9
src/smk_playground/cli.py
View file

@ -19,5 +19,14 @@ def status() -> None:
typer.echo(f"{'FOUND' if path.exists() else 'MISSING'} {path}") typer.echo(f"{'FOUND' if path.exists() else 'MISSING'} {path}")
@app.command("milestone1")
def milestone1(root: Path | None = None) -> None:
"""Build manifest-bound source Parquets and Milestone 1 audits."""
from smk_playground.ingest.milestone1 import run
summary = run((root or Path.cwd()).resolve())
typer.echo(f"Milestone 1 complete: {summary['source_rows']}")
if __name__ == "__main__": if __name__ == "__main__":
app() app()

515
src/smk_playground/ingest/milestone1.py Normal file
View file

@ -0,0 +1,515 @@
"""Milestone 1 streaming ingest and source-audit artifact generation."""
from __future__ import annotations
import argparse
import hashlib
import json
import statistics
from collections import Counter, defaultdict
from pathlib import Path
from typing import Any
import pyarrow as pa
import pyarrow.parquet as pq
from .source import (
HEX_COLOR_RE,
iter_source_records,
logical_json,
logical_sha256,
normalize_dimension_axis,
normalize_dimension_unit,
normalized_dimensions,
parse_strict_number,
source_row_id,
)
VISUAL_FIELDS = ("brightness", "contrast", "saturation", "entropy", "colortemp")
SOURCE_FIELDS = (
"id", "object_number", "created", "modified", "public_domain", "rights", "has_image",
"image_width", "image_height", "image_mime_type", "image_iiif_id", "image_iiif_info",
"image_thumbnail", "image_native", "colors", "brightness", "contrast", "saturation",
"entropy", "colortemp", "dimensions", "titles", "object_names", "classification",
"classifications",
)
MANIFEST_HASH_FIELDS = {"manifest_version", "generated_at_utc", "reconnaissance_tool", "sources"}
SOURCE_SCHEMA = pa.schema([
("source_row_id", pa.string()),
("source_language", pa.string()),
("source_path", pa.string()),
("source_file_sha256", pa.string()),
("source_manifest_sha256", pa.string()),
("smk_source_id", pa.string()),
("object_number", pa.string()),
("created", pa.string()),
("modified", pa.string()),
("public_domain", pa.bool_()),
("rights", pa.string()),
("has_image", pa.bool_()),
("image_width", pa.int64()),
("image_height", pa.int64()),
("image_iiif_id", pa.string()),
("image_iiif_info", pa.string()),
("image_thumbnail", pa.string()),
("image_native", pa.string()),
("colors_json", pa.string()),
("dimensions_json", pa.string()),
("titles_json", pa.string()),
("object_names_json", pa.string()),
("visual_json", pa.string()),
("normalized_dimensions_json", pa.string()),
("record_logical_sha256", pa.string()),
("source_record_json", pa.string()),
])
def _optional_string(value: Any) -> str | None:
return value if isinstance(value, str) else None
def _optional_int(value: Any) -> int | None:
return value if isinstance(value, int) and not isinstance(value, bool) else None
def _optional_bool(value: Any) -> bool | None:
return value if isinstance(value, bool) else None
def _json_or_none(value: Any) -> str | None:
return logical_json(value) if value is not None else None
def _manifest_sha256(manifest: dict[str, Any]) -> str:
stable = {key: manifest[key] for key in sorted(MANIFEST_HASH_FIELDS) if key in manifest}
return hashlib.sha256(logical_json(stable).encode()).hexdigest()
def _source_info(manifest: dict[str, Any], language: str) -> dict[str, Any]:
return next(source for source in manifest["sources"] if source["language"] == language)
class AuditState:
def __init__(self, language: str) -> None:
self.language = language
self.row_ids: set[str] = set()
self.smk_ids: set[str] = set()
self.object_counts: Counter[str] = Counter()
self.dimension_vocab: Counter[tuple[str, str, str, str, str, str]] = Counter()
self.dimension_values: dict[tuple[str, str, str], set[str]] = defaultdict(set)
self.dimension_issues: Counter[str] = Counter()
self.dimension_duplicate_measurements = 0
self.dimension_conflicting_measurements = 0
self.palette_lengths: Counter[int] = Counter()
self.palette_missing = 0
self.palette_invalid = 0
self.palette_duplicate_records = 0
self.palette_invalid_examples: list[dict[str, str]] = []
self.visual_values: dict[str, list[float]] = {field: [] for field in VISUAL_FIELDS}
self.visual_types: dict[str, Counter[str]] = {field: Counter() for field in VISUAL_FIELDS}
self.visual_malformed: dict[str, list[dict[str, Any]]] = {field: [] for field in VISUAL_FIELDS}
self.rights: Counter[str] = Counter()
self.public_domain: Counter[str] = Counter()
self.has_image: Counter[str] = Counter()
self.image_url_presence: Counter[str] = Counter()
def observe(self, record: dict[str, Any], row_id: str) -> None:
if row_id in self.row_ids:
raise ValueError(f"Non-unique source_row_id: {row_id}")
self.row_ids.add(row_id)
smk_id = record.get("id")
if isinstance(smk_id, str) and smk_id:
if smk_id in self.smk_ids:
raise ValueError(f"SMK id is not unique in {self.language}: {smk_id}")
self.smk_ids.add(smk_id)
object_number = record.get("object_number")
if isinstance(object_number, str):
self.object_counts[object_number] += 1
self._observe_dimensions(record, row_id)
self._observe_palette(record)
self._observe_visuals(record)
rights = record.get("rights")
self.rights[str(rights) if rights is not None else "<missing>"] += 1
self.public_domain[str(record.get("public_domain"))] += 1
self.has_image[str(record.get("has_image"))] += 1
has_public_url = any(record.get(key) for key in ("image_native", "image_thumbnail", "image_iiif_id"))
self.image_url_presence[str(bool(has_public_url))] += 1
def _observe_dimensions(self, record: dict[str, Any], row_id: str) -> None:
dimensions = record.get("dimensions")
if not isinstance(dimensions, list):
self.dimension_issues["dimensions_not_array_or_missing"] += 1
return
seen: Counter[tuple[str, str, str, str]] = Counter()
grouped: dict[tuple[str, str, str], set[str]] = defaultdict(set)
for item in dimensions:
if not isinstance(item, dict):
self.dimension_issues["dimension_item_not_object"] += 1
continue
part = str(item.get("part") or "")
dim_type = str(item.get("type") or "")
unit = str(item.get("unit") or "")
raw_value = item.get("value")
value_text = "" if raw_value is None else str(raw_value)
_, number_status = parse_strict_number(raw_value)
axis = normalize_dimension_axis(dim_type) or "<unmapped>"
normalized_unit = normalize_dimension_unit(unit) or "<unmapped>"
self.dimension_vocab[(part, dim_type, unit, axis, normalized_unit, number_status)] += 1
self.dimension_values[(part, dim_type, unit)].add(value_text)
if not part: self.dimension_issues["missing_part"] += 1
if not dim_type: self.dimension_issues["missing_type"] += 1
if not unit: self.dimension_issues["missing_unit"] += 1
if number_status not in {"numeric", "numeric_text"}:
self.dimension_issues[f"value_{number_status}"] += 1
key = (part, dim_type, unit, value_text)
seen[key] += 1
grouped[(part, dim_type, unit)].add(value_text)
self.dimension_duplicate_measurements += sum(count - 1 for count in seen.values() if count > 1)
self.dimension_conflicting_measurements += sum(1 for values in grouped.values() if len(values) > 1)
def _observe_palette(self, record: dict[str, Any]) -> None:
colors = record.get("colors")
if not isinstance(colors, list):
self.palette_missing += 1
return
self.palette_lengths[len(colors)] += 1
normalized: list[str] = []
for color in colors:
if isinstance(color, str) and HEX_COLOR_RE.fullmatch(color):
normalized.append(color.casefold())
else:
self.palette_invalid += 1
if len(self.palette_invalid_examples) < 20:
self.palette_invalid_examples.append({
"object_number": str(record.get("object_number")), "value": repr(color),
})
if len(set(normalized)) < len(normalized):
self.palette_duplicate_records += 1
def _observe_visuals(self, record: dict[str, Any]) -> None:
for field in VISUAL_FIELDS:
raw = record.get(field)
type_name = type(raw).__name__ if raw is not None else "missing"
self.visual_types[field][type_name] += 1
number, status = parse_strict_number(raw)
if number is not None:
self.visual_values[field].append(number)
elif raw is not None and len(self.visual_malformed[field]) < 30:
self.visual_malformed[field].append({
"object_number": record.get("object_number"), "value": raw, "status": status,
})
def _source_row(
record: dict[str, Any], language: str, source: dict[str, Any], manifest_sha: str,
) -> dict[str, Any]:
visual = {field: record.get(field) for field in VISUAL_FIELDS if field in record}
return {
"source_row_id": source_row_id(language, record),
"source_language": language,
"source_path": source["path"],
"source_file_sha256": source["sha256"],
"source_manifest_sha256": manifest_sha,
"smk_source_id": _optional_string(record.get("id")),
"object_number": _optional_string(record.get("object_number")),
"created": _optional_string(record.get("created")),
"modified": _optional_string(record.get("modified")),
"public_domain": _optional_bool(record.get("public_domain")),
"rights": _optional_string(record.get("rights")),
"has_image": _optional_bool(record.get("has_image")),
"image_width": _optional_int(record.get("image_width")),
"image_height": _optional_int(record.get("image_height")),
"image_iiif_id": _optional_string(record.get("image_iiif_id")),
"image_iiif_info": _optional_string(record.get("image_iiif_info")),
"image_thumbnail": _optional_string(record.get("image_thumbnail")),
"image_native": _optional_string(record.get("image_native")),
"colors_json": _json_or_none(record.get("colors")),
"dimensions_json": _json_or_none(record.get("dimensions")),
"titles_json": _json_or_none(record.get("titles")),
"object_names_json": _json_or_none(record.get("object_names")),
"visual_json": logical_json(visual),
"normalized_dimensions_json": logical_json(normalized_dimensions(record)),
"record_logical_sha256": logical_sha256(record),
"source_record_json": logical_json(record),
}
def ingest_source(
root: Path, language: str, manifest: dict[str, Any], batch_size: int = 1_000,
) -> AuditState:
source = _source_info(manifest, language)
source_path = root / source["path"]
output_path = root / "data" / "interim" / f"source_{language}.parquet"
output_path.parent.mkdir(parents=True, exist_ok=True)
manifest_sha = _manifest_sha256(manifest)
metadata = {
b"smk.source_manifest_sha256": manifest_sha.encode(),
b"smk.source_file_sha256": source["sha256"].encode(),
b"smk.source_language": language.encode(),
b"smk.logical_lossless": b"source_record_json preserves the complete parsed JSON record",
}
writer = pq.ParquetWriter(output_path, SOURCE_SCHEMA.with_metadata(metadata), compression="zstd")
state = AuditState(language)
batch: list[dict[str, Any]] = []
try:
for record in iter_source_records(source_path):
row = _source_row(record, language, source, manifest_sha)
state.observe(record, row["source_row_id"])
batch.append(row)
if len(batch) >= batch_size:
writer.write_table(pa.Table.from_pylist(batch, schema=SOURCE_SCHEMA))
batch.clear()
if batch:
writer.write_table(pa.Table.from_pylist(batch, schema=SOURCE_SCHEMA))
finally:
writer.close()
if len(state.row_ids) != source["record_count"]:
raise ValueError(f"{language} row count differs from source manifest")
return state
def _read_rows(path: Path, columns: list[str] | None = None) -> list[dict[str, Any]]:
return pq.read_table(path, columns=columns).to_pylist()
def write_duplicate_report(root: Path, states: dict[str, AuditState]) -> dict[str, Any]:
rows: list[dict[str, Any]] = []
taxonomy = Counter()
for language, state in states.items():
duplicate_ids = {key for key, count in state.object_counts.items() if count > 1}
source_rows = _read_rows(root / "data" / "interim" / f"source_{language}.parquet")
groups: dict[str, list[dict[str, Any]]] = defaultdict(list)
for row in source_rows:
if row["object_number"] in duplicate_ids:
groups[row["object_number"]].append(row)
for object_number, group in sorted(groups.items()):
hashes = {row["record_logical_sha256"] for row in group}
records = [json.loads(row["source_record_json"]) for row in group]
differing = [
field for field in sorted(set().union(*(record.keys() for record in records)))
if len({logical_json(record.get(field)) for record in records}) > 1
]
if len(hashes) == 1:
category = "exact_logical_duplicate"
elif all(record.get("id") == records[0].get("id") for record in records):
category = "same_source_id_changed_record"
elif any("verso" in str(record.get("object_number", "")).casefold() for record in records):
category = "possible_alternate_or_subrecord"
elif all(record.get("id") for record in records):
category = "multiple_source_records_same_object_number"
else:
category = "unresolved"
taxonomy[category] += 1
for row, record in zip(group, records, strict=True):
rows.append({
"language": language,
"object_number": object_number,
"occurrence_count": len(group),
"source_row_id": row["source_row_id"],
"smk_source_id": row["smk_source_id"],
"record_logical_sha256": row["record_logical_sha256"],
"logically_identical_group": len(hashes) == 1,
"byte_identical": False if len(hashes) > 1 else None,
"byte_identity_note": (
"Logical content differs, therefore source bytes cannot be identical"
if len(hashes) > 1 else
"Logical content matches; source byte spans were not retained"
),
"differing_fields_json": logical_json(differing),
"classification": category,
"modified": row["modified"],
"image_identifiers_json": logical_json({
key: record.get(key) for key in
("image_iiif_id", "image_iiif_info", "image_native", "image_thumbnail")
}),
"titles_json": row["titles_json"],
"object_type_json": logical_json({
"object_names": record.get("object_names"),
"classification": record.get("classification"),
"classifications": record.get("classifications"),
}),
"rights": row["rights"],
"dimensions_json": row["dimensions_json"],
"colors_json": row["colors_json"],
"visual_json": row["visual_json"],
})
table = pa.Table.from_pylist(rows)
pq.write_table(table, root / "data" / "reports" / "duplicate-object-numbers.parquet", compression="zstd")
return {"groups_by_taxonomy": dict(taxonomy), "rows": len(rows)}
def write_dimension_report(root: Path, states: dict[str, AuditState]) -> dict[str, Any]:
rows: list[dict[str, Any]] = []
summary: dict[str, Any] = {}
for language, state in states.items():
for (part, dim_type, unit, axis, normalized_unit, value_status), count in sorted(state.dimension_vocab.items()):
values = state.dimension_values[(part, dim_type, unit)]
rows.append({
"language": language, "part": part, "type": dim_type, "unit": unit,
"proposed_axis": axis, "proposed_unit": normalized_unit,
"value_status": value_status, "count": count, "distinct_value_count": len(values),
})
summary[language] = {
"issues": dict(state.dimension_issues),
"duplicate_measurements_beyond_first": state.dimension_duplicate_measurements,
"conflicting_part_type_unit_groups": state.dimension_conflicting_measurements,
"vocabulary_rows": sum(1 for row in rows if row["language"] == language),
}
pq.write_table(pa.Table.from_pylist(rows), root / "data" / "reports" / "dimension-vocabulary.parquet", compression="zstd")
return summary
def _comparison_value(row: dict[str, Any], field: str) -> Any:
if field == "colors": return json.loads(row["colors_json"]) if row["colors_json"] else None
if field == "dimensions": return json.loads(row["normalized_dimensions_json"])
if field in VISUAL_FIELDS:
return json.loads(row["visual_json"]).get(field)
return row.get(field)
def write_disagreement_report(root: Path, states: dict[str, AuditState]) -> dict[str, Any]:
columns = [
"source_row_id", "object_number", "rights", "public_domain", "has_image", "image_width",
"image_height", "image_iiif_id", "image_iiif_info", "image_thumbnail", "image_native",
"colors_json", "visual_json", "normalized_dimensions_json",
]
by_language: dict[str, dict[str, list[dict[str, Any]]]] = {}
for language in ("da", "en"):
grouped: dict[str, list[dict[str, Any]]] = defaultdict(list)
for row in _read_rows(root / "data" / "interim" / f"source_{language}.parquet", columns):
grouped[row["object_number"]].append(row)
by_language[language] = grouped
fields = (
"rights", "public_domain", "has_image", "image_width", "image_height", "image_iiif_id",
"image_iiif_info", "image_thumbnail", "image_native", "colors", *VISUAL_FIELDS, "dimensions",
)
rows: list[dict[str, Any]] = []
classifications = Counter()
overlap = sorted(set(by_language["da"]) & set(by_language["en"]))
for object_number in overlap:
da_group = by_language["da"][object_number]
en_group = by_language["en"][object_number]
if len(da_group) != 1 or len(en_group) != 1:
classifications["duplicate_identity_requires_resolution"] += 1
rows.append({
"object_number": object_number, "field": "<identity>",
"classification": "duplicate_identity_requires_resolution",
"da_value_json": logical_json([row["source_row_id"] for row in da_group]),
"en_value_json": logical_json([row["source_row_id"] for row in en_group]),
"da_source_row_id": None, "en_source_row_id": None,
})
continue
da_row, en_row = da_group[0], en_group[0]
for field in fields:
da_value = _comparison_value(da_row, field)
en_value = _comparison_value(en_row, field)
if logical_json(da_value) == logical_json(en_value):
continue
if da_value is None and en_value is not None:
category = "missing_in_da"
elif en_value is None and da_value is not None:
category = "missing_in_en"
elif field == "rights":
category = "localized_or_license_text_difference"
elif field == "dimensions":
category = "normalized_dimension_difference"
elif field.startswith("image_") or field == "has_image":
category = "image_metadata_difference"
else:
category = "value_difference"
classifications[f"{field}:{category}"] += 1
rows.append({
"object_number": object_number, "field": field, "classification": category,
"da_value_json": logical_json(da_value), "en_value_json": logical_json(en_value),
"da_source_row_id": da_row["source_row_id"],
"en_source_row_id": en_row["source_row_id"],
})
pq.write_table(pa.Table.from_pylist(rows), root / "data" / "reports" / "source-disagreements.parquet", compression="zstd")
return {"overlap": len(overlap), "disagreement_rows": len(rows), "classifications": dict(classifications)}
def _percentiles(values: list[float]) -> dict[str, float] | None:
if not values: return None
ordered = sorted(values)
def quantile(p: float) -> float:
position = (len(ordered) - 1) * p
low = int(position)
high = min(low + 1, len(ordered) - 1)
fraction = position - low
return ordered[low] * (1 - fraction) + ordered[high] * fraction
return {
"min": ordered[0], "p01": quantile(0.01), "p05": quantile(0.05),
"p25": quantile(0.25), "median": statistics.median(ordered), "p75": quantile(0.75),
"p95": quantile(0.95), "p99": quantile(0.99), "max": ordered[-1],
}
def write_profiles(root: Path, states: dict[str, AuditState], disagreements: dict[str, Any]) -> None:
palette_visual: dict[str, Any] = {"languages": {}}
rights: dict[str, Any] = {"languages": {}}
for language, state in states.items():
visual = {}
for field in VISUAL_FIELDS:
visual[field] = {
"datatype_counts": dict(state.visual_types[field]),
"valid_numeric_count": len(state.visual_values[field]),
"statistics": _percentiles(state.visual_values[field]),
"malformed_examples": state.visual_malformed[field],
}
palette_visual["languages"][language] = {
"colors_missing_or_non_array": state.palette_missing,
"palette_length_distribution": dict(sorted(state.palette_lengths.items())),
"invalid_color_values": state.palette_invalid,
"invalid_color_examples": state.palette_invalid_examples,
"records_with_duplicate_colors": state.palette_duplicate_records,
"visual_fields": visual,
}
rights["languages"][language] = {
"rights_value_counts": dict(state.rights.most_common()),
"public_domain_counts": dict(state.public_domain),
"has_image_counts": dict(state.has_image),
"public_image_url_presence": dict(state.image_url_presence),
}
palette_visual["disagreement_classifications"] = disagreements["classifications"]
report_dir = root / "data" / "reports"
(report_dir / "palette-visual-profile.json").write_text(
json.dumps(palette_visual, ensure_ascii=False, indent=2), encoding="utf-8",
)
(report_dir / "rights-public-domain-profile.json").write_text(
json.dumps(rights, ensure_ascii=False, indent=2), encoding="utf-8",
)
def run(root: Path) -> dict[str, Any]:
manifest_path = root / "data" / "processed" / "source-manifest.json"
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
states = {language: ingest_source(root, language, manifest) for language in ("da", "en")}
duplicate_summary = write_duplicate_report(root, states)
dimension_summary = write_dimension_report(root, states)
disagreement_summary = write_disagreement_report(root, states)
write_profiles(root, states, disagreement_summary)
summary = {
"source_manifest_sha256": _manifest_sha256(manifest),
"source_rows": {language: len(state.row_ids) for language, state in states.items()},
"unique_smk_source_ids": {language: len(state.smk_ids) for language, state in states.items()},
"duplicates": duplicate_summary,
"dimensions": dimension_summary,
"disagreements": disagreement_summary,
}
(root / "data" / "reports" / "milestone1-summary.json").write_text(
json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8",
)
return summary
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--root", type=Path, default=Path.cwd())
args = parser.parse_args()
print(json.dumps(run(args.root.resolve()), indent=2))
if __name__ == "__main__":
main()

134
src/smk_playground/ingest/source.py Normal file
View file

@ -0,0 +1,134 @@
"""Bounded-memory loading and stable source-row helpers for SMK snapshots."""
from __future__ import annotations
import hashlib
import json
import math
import re
from collections.abc import Iterator, Mapping
from decimal import Decimal
from pathlib import Path
from typing import Any
import ijson
HEX_COLOR_RE = re.compile(r"^#[0-9a-fA-F]{6}$")
STRICT_NUMBER_RE = re.compile(r"^[+-]?(?:\d+(?:\.\d*)?|\.\d+)(?:[eE][+-]?\d+)?$")
def iter_source_records(path: Path) -> Iterator[dict[str, Any]]:
"""Yield objects from the observed top-level array without loading the file."""
with path.open("rb") as stream:
for record in ijson.items(stream, "item", use_float=False):
if not isinstance(record, dict):
raise TypeError(f"Expected object at $[] in {path}")
yield _normalize_decimal_numbers(record)
def _normalize_decimal_numbers(value: Any) -> Any:
"""Convert ijson Decimal values while retaining JSON number semantics."""
if isinstance(value, Decimal):
return int(value) if value == value.to_integral_value() else float(value)
if isinstance(value, list):
return [_normalize_decimal_numbers(item) for item in value]
if isinstance(value, dict):
return {key: _normalize_decimal_numbers(item) for key, item in value.items()}
return value
def logical_json(value: Any) -> str:
"""Return deterministic, logically lossless JSON for storage and hashing."""
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
def logical_sha256(value: Any) -> str:
return hashlib.sha256(logical_json(value).encode("utf-8")).hexdigest()
def source_row_id(language: str, record: Mapping[str, Any]) -> str:
"""Build a stable row ID from SMK's source id; never use array position."""
smk_id = record.get("id")
if isinstance(smk_id, str) and smk_id.strip():
return f"smk:{language}:{smk_id.strip()}"
stable_fallback = {
"language": language,
"object_number": record.get("object_number"),
"record": record,
}
return f"sha256:{logical_sha256(stable_fallback)}"
def parse_strict_number(value: Any) -> tuple[float | None, str]:
"""Parse a finite number without silently accepting malformed text."""
if isinstance(value, bool) or value is None:
return None, "missing" if value is None else "invalid_type"
if isinstance(value, (int, float)):
number = float(value)
return (number, "numeric") if math.isfinite(number) else (None, "non_finite")
if isinstance(value, str):
text = value.strip()
if not text:
return None, "blank"
if not STRICT_NUMBER_RE.fullmatch(text):
return None, "malformed_text"
number = float(text)
return (number, "numeric_text") if math.isfinite(number) else (None, "non_finite")
return None, "invalid_type"
def normalize_dimension_axis(value: Any) -> str | None:
if not isinstance(value, str):
return None
token = value.strip().casefold()
mapping = {
"bredde": "width",
"width": "width",
"højde": "height",
"hojde": "height",
"height": "height",
"dybde": "depth",
"depth": "depth",
"diameter": "diameter",
"længde": "length",
"laengde": "length",
"length": "length",
}
return mapping.get(token)
def normalize_dimension_unit(value: Any) -> str | None:
if not isinstance(value, str):
return None
token = value.strip().casefold().replace(".", "")
mapping = {
"millimeter": "mm",
"millimetre": "mm",
"mm": "mm",
"centimeter": "cm",
"centimetre": "cm",
"cm": "cm",
"meter": "m",
"metre": "m",
"m": "m",
}
return mapping.get(token)
def normalized_dimensions(record: Mapping[str, Any]) -> list[tuple[str, str, str, float]]:
"""Create an audit-only localized comparison signature, not canonical dimensions."""
output: list[tuple[str, str, str, float]] = []
dimensions = record.get("dimensions")
if not isinstance(dimensions, list):
return output
for item in dimensions:
if not isinstance(item, dict):
continue
axis = normalize_dimension_axis(item.get("type"))
unit = normalize_dimension_unit(item.get("unit"))
number, status = parse_strict_number(item.get("value"))
if axis and unit and number is not None and status in {"numeric", "numeric_text"}:
part = str(item.get("part") or "").strip().casefold()
output.append((part, axis, unit, number))
return sorted(output)

4
tests/fixtures/source_array.json vendored Normal file
View file

@ -0,0 +1,4 @@
[
{"id":"1_object","object_number":"A1","dimensions":[{"part":"Netto","type":"højde","unit":"millimeter","value":"10"}],"colortemp":1.5},
{"id":"2_object","object_number":"A1","dimensions":[{"part":"Netto","type":"width","unit":"mm","value":"bad"}],"colortemp":"2.5"}
]

69
tests/test_source_ingest.py Normal file
View file

@ -0,0 +1,69 @@
import json
from pathlib import Path
import pyarrow.parquet as pq
from smk_playground.ingest.milestone1 import ingest_source
from smk_playground.ingest.source import (
iter_source_records,
normalize_dimension_axis,
parse_strict_number,
source_row_id,
)
FIXTURE = Path(__file__).parent / "fixtures" / "source_array.json"
def test_streaming_parser_preserves_duplicates() -> None:
records = list(iter_source_records(FIXTURE))
assert [record["object_number"] for record in records] == ["A1", "A1"]
def test_source_row_id_is_deterministic_and_not_positional() -> None:
record = {"id": "117_object", "object_number": "A1"}
assert source_row_id("da", record) == "smk:da:117_object"
assert source_row_id("da", record) == source_row_id("da", dict(reversed(record.items())))
assert source_row_id("en", record) != source_row_id("da", record)
def test_dimension_vocabulary_mapping_is_explicit() -> None:
assert normalize_dimension_axis("højde") == "height"
assert normalize_dimension_axis("width") == "width"
assert normalize_dimension_axis("omkreds") is None
def test_malformed_numeric_handling_is_not_silent() -> None:
assert parse_strict_number("12.5") == (12.5, "numeric_text")
assert parse_strict_number("12,5") == (None, "malformed_text")
assert parse_strict_number(None) == (None, "missing")
def test_streaming_parser_preserves_large_json_integers(tmp_path: Path) -> None:
source = tmp_path / "large-number.json"
source.write_text('[{"id":"x","colortemp":2673963600000000000000000000}]')
record = next(iter_source_records(source))
assert record["colortemp"] == 2673963600000000000000000000
def test_parquet_round_trip_preserves_complete_records_and_duplicates(tmp_path: Path) -> None:
raw = tmp_path / "data" / "raw"
processed = tmp_path / "data" / "processed"
raw.mkdir(parents=True)
processed.mkdir(parents=True)
source_path = raw / "fixture.json"
source_path.write_bytes(FIXTURE.read_bytes())
manifest = {
"manifest_version": "test",
"sources": [{
"language": "da", "path": "data/raw/fixture.json", "sha256": "fixture",
"record_count": 2,
}],
}
state = ingest_source(tmp_path, "da", manifest, batch_size=1)
rows = pq.read_table(tmp_path / "data" / "interim" / "source_da.parquet").to_pylist()
assert state.object_counts["A1"] == 2
assert len(rows) == 2
assert json.loads(rows[0]["source_record_json"])["dimensions"][0]["value"] == "10"
assert rows[0]["source_row_id"] != rows[1]["source_row_id"]