#!/usr/bin/env python3 """Build citations for official statistics, with SDMX provenance handling. Three subcommands: cite Assemble a citation from source fields and render the canonical source block — one Source, one Licence, one Note. Also renders the legacy footnote and reference-list forms. Validates completeness and refuses to publish a block whose formal name or selection is unresolved. flag Resolve a data-quality flag. Agency publication symbols come from a curated table; SDMX OBS_STATUS/CONF_STATUS are resolved against the code list the DSD names, supplied via --codelist. urn Construct or check an SDMX URN from an AGENCY:ID(VERSION) triplet. The naming contract, in one line: a title and a selection are the provider's words, a response format is not part of either, and neither is ever invented here to make validation pass. See references/agency-formats.md and references/units-and-tiers.md, which ship alongside this script. WHAT THE PROFILE 4 GATE PROVES, AND WHAT IT DOES NOT. It proves the block is SHAPED correctly: a title and a selection are present with explicit availability states, every selected member carries a role with named evidence, a licence label is paired with an absolute terms URL, the vintage is present, the link text is a name rather than a mechanism, and the codes plausibly match the query. It does NOT prove those values came from the provider — this script makes no network calls, so it cannot fetch the DSD and compare. Authenticity rests on the agent having resolved provider metadata before populating the fields, and on setting resource_title_status / selection_status to "unavailable" when it could not. A caller determined to fabricate a well-shaped citation can still do so; the gate is a guard rail against carelessness, not an authentication. No third-party dependencies. Python 3.8+. Examples -------- python3 sdmx_cite.py cite \\ --agency "Australian Bureau of Statistics" \\ --agency-id ABS --dataflow-id IIP --version 1.0.0 \\ --reference-period "1988-Q3-2026-Q1" \\ --resource-title "International Investment Position" \\ --resource-kind dataflow \\ --selected-member-json '{"dimension_id": "MEASURE", "dimension_label": "Measure", "code": "6", "member_label": "Position at end of period", "citation_role": "identity", "role_evidence": "default_identity"}' \\ --query-url "https://data.api.abs.gov.au/rest/data/IIP/6.903A...Q" \\ --response-format CSV \\ --licence-label "CC BY 4.0 International (ABS terms)" \\ --licence-url "https://www.abs.gov.au/website-privacy-copyright-and-disclaimer" \\ --extracted-at 2026-08-07 --style abs python3 sdmx_cite.py cite --json citation.json --format json python3 sdmx_cite.py flag --source eurostat --code ":c" python3 sdmx_cite.py flag --code R --codelist obs_status.json python3 sdmx_cite.py urn --triplet "ESTAT:ISOC_CI_ID_H(1.0)" The --json file holds either one citation object or {"citations": [...]}, which is the only way to render a multi-source block. """ from __future__ import annotations import argparse import json import re import sys from datetime import datetime, timezone from typing import Any, Dict, List, Optional, Tuple from urllib.parse import parse_qsl, unquote, urlencode, urlsplit, urlunsplit # -------------------------------------------------------------------------- # Flag code sets. Deliberately per-source: the same character means different # things in different systems, so these are never merged into one table. # -------------------------------------------------------------------------- FLAGS: Dict[str, Dict[str, str]] = { "eurostat": { ":": "not available", ":c": "confidential", ":z": "not applicable", ":n": "not significant", "0": "real zero", "0n": "less than half the final digit shown, greater than real zero", "p": "provisional (attaches only to a value)", "e": "estimated (attaches only to a value)", "f": "forecast (attaches only to a value)", "b": "break in series", "|": "break in series", "d": "definition differs — see metadata", "u": "low reliability", "a": "below reliability limit — suppress, publish a dot (EU-LFS)", "z": "not applicable (only meaningful combined with ':')", "c": "confidential (only meaningful combined with ':')", "n": "not significant (only meaningful combined with ':')", }, "abs": { "n.a.": "not available", ". .": "not applicable", "—": "nil or rounded to zero", "p": "provisional (older publications)", "n.p.": "not for publication", "^": "RSE 10-25% — use with caution", "*": "RSE 25-50% — use with caution (older pubs: high standard errors)", "**": "RSE > 50% — too unreliable for general use", "n.e.c.": "not elsewhere classified", "n.e.i.": "not elsewhere included", "n.e.s.": "not elsewhere specified", "n.f.d.": "not further defined", "n.y.a.": "not yet available", }, "statcan": { ".": "not available for any reference period", "..": "not available for a specific reference period", "...": "not applicable", "0": "true zero", "0s": "value rounded to zero where rounding is meaningful", "p": "provisional", "r": "revised", "x": "suppressed under the Statistics Act", "E": "use with caution — CV > 16.5% and <= 33.3%", "F": "too unreliable to publish — CV > 33.3% or sample size < 10", "A": "excellent quality", "B": "very good quality", "C": "good quality (yellow triangle C denotes a high-level correction)", "D": "acceptable quality", }, } # --------------------------------------------------------------------------- # SDMX flag code lists are RESOLVED AT RUNTIME, not carried here. # # `OBS_STATUS` and `CONF_STATUS` are DSD-declared attributes: the data structure # says which codelist decodes them, and providers routinely point at their own # rather than the SDMX cross-domain list. Eurostat re-coded three of its # observation-status flags on 27 January 2025 and maintains its own `Obs_status` # list; a table shipped in this file cannot know either fact. Worse, it answers # anyway — a hardcoded cross-domain copy silently glossed `R` as "revised" when # CL_OBS_STATUS 2.3 defines it as "excludes one or more subcategories", which # changes what the number counts. # # So there is nothing to hardcode. Supply the codelist the DSD points at, via # `--codelist`, and this resolves against that. Without it the answer is # UNRESOLVED — see cmd_flag. The agency tables above stay, because they are # publication conventions in rendered tables (Eurostat's `:c`, ABS's `n.p.`, # StatCan's CV grades) and appear in no DSD. SDMX_ATTRIBUTES = { "sdmx": "OBS_STATUS", "sdmx_conf": "CONF_STATUS", } def load_codelist(payload) -> Tuple[Dict[str, str], str]: """(codes, provenance) from a codelist document. Accepts the shape an origin/graph lookup naturally produces:: {"agency_id": "ESTAT", "id": "OBS_STATUS", "version": "1.0", "codes": {"A": "Normal value", "p": "Provisional"}} and SDMX-JSON structure messages as returned by a registry or provider:: {"data": {"codelists": [{"id": ..., "agency": ..., "version": ..., "codes": [{"id": "A", "name": "Normal value"}]}]}} """ doc = payload if isinstance(doc, dict) and "data" in doc and isinstance(doc["data"], dict): doc = doc["data"] if isinstance(doc, dict) and isinstance(doc.get("codelists"), list): if not doc["codelists"]: raise ValueError("codelist document contains no codelists") doc = doc["codelists"][0] if not isinstance(doc, dict): raise ValueError("codelist document is not an object") raw = doc.get("codes") if isinstance(raw, dict): codes = {str(k): str(v) for k, v in raw.items()} elif isinstance(raw, list): codes = {} for entry in raw: if not isinstance(entry, dict) or "id" not in entry: raise ValueError("each code needs an 'id'") name = entry.get("name") or entry.get("names", {}).get("en") if name is None: raise ValueError(f"code {entry['id']!r} has no name") codes[str(entry["id"])] = str(name) else: raise ValueError("codelist document has no 'codes'") if not codes: raise ValueError("codelist document contains no codes") agency = doc.get("agency_id") or doc.get("agency") or doc.get("agencyID") cid = doc.get("id") version = doc.get("version") parts = [p for p in (agency, cid) if p] provenance = ":".join(parts) if parts else "supplied codelist" if version: provenance += f"({version})" return codes, provenance URN_CLASSES = { "dataflow": "datastructure.Dataflow", "datastructure": "datastructure.DataStructure", "dsd": "datastructure.DataStructure", "codelist": "codelist.Codelist", "conceptscheme": "conceptscheme.ConceptScheme", "categoryscheme": "categoryscheme.CategoryScheme", "provisionagreement": "registry.ProvisionAgreement", } # Real SDMX ids contain @ and . (e.g. OECD.SDD.NAD:DSD_NAAG@DF_NAAG_I) TRIPLET_RE = re.compile(r"^([A-Za-z0-9_.\-]+):([A-Za-z0-9_.@\-]+)\((\d+(?:\.\d+)*)\)$") # -------------------------------------------------------------------------- # The citation-identity vocabulary. These spellings are the shared contract in # references/units-and-tiers.md and are deliberately identical end to end — a # renderer that has to guess at a synonym eventually guesses wrong. # -------------------------------------------------------------------------- CITATION_ROLES = ("identity", "interpretation", "neutral_scope") RESOURCE_KINDS = ("resource", "dataflow") TITLE_STATUSES = ("verified", "unavailable") SELECTION_STATUSES = ("verified", "not_applicable", "unavailable") SELECTION_BASES = ("whole_resource", "subset") # The one evidence value that is a claim about nothing: "no metadata justified # a non-default role, so this member stayed identity". DEFAULT_IDENTITY = "default_identity" INTERPRETATION_EVIDENCE = "concept:" NEUTRAL_SCOPE_EVIDENCE = ("provider_aggregate:", "codelist:") SELECTION_JOINER = " — " ENTRY_JOINER = "; " # Link text names the data, never the mechanism. These are the labels that # show up when an agent reaches for a description of the transport instead of # the provider's own words. MECHANISM_LABELS = ( "exact query", "exact data api query", "data api query", "api query", "api call", "query url", "data url", "query link", "click here", "link", "download", "source data", ) # Response formats, for the "format embedded in the formal name" check. RESPONSE_FORMAT_TOKENS = ( "csv", "sdmx-csv", "json", "sdmx-json", "xml", "sdmx-ml", "xlsx", "xls", "tsv", "parquet", ) # UNIT_MULT is a power of ten. Only the conventional English scale words are # mapped; an unmapped exponent renders as the raw power rather than a coined # word, because "10^7 dollars" is honest and "ten-millions" is not idiomatic. UNIT_MULT_WORDS = { "0": "", "3": "thousands", "6": "millions", "9": "billions", "12": "trillions", } # Bound concept identities the ABS Profile 4 note formatter composes. Matched on # the concept named in role_evidence, never on the dimension's spelling. ADJUSTMENT_CONCEPTS = ("ADJUSTMENT", "SEASONAL_ADJUST", "ADJUSTMENT_TYPE") FREQ_CONCEPTS = ("FREQ", "FREQUENCY") UNIT_CONCEPTS = ("UNIT_MEASURE", "CURRENCY", "UNIT") # -------------------------------------------------------------------------- # URLs # -------------------------------------------------------------------------- def is_http_url(url: Optional[str]) -> bool: if not url: return False parts = urlsplit(url.strip()) return parts.scheme.lower() in ("http", "https") and bool(parts.netloc) def normalize_query_url(url: Optional[str]) -> str: """Normalised query identity: absolute, stable parameter order, no fragment. Used for the dedupe key only. The rendered link keeps the URL exactly as it was supplied — normalising what the reader sees would silently rewrite a provider's own canonical form. """ if not url: return "" parts = urlsplit(url.strip()) query = urlencode(sorted(parse_qsl(parts.query, keep_blank_values=True))) return urlunsplit( (parts.scheme.lower(), parts.netloc.lower(), parts.path, query, "")) def _squash(text: str) -> str: return " ".join(text.split()) def _pluralise(label: str) -> str: """Narrow, deliberate: unit labels are singular nouns like 'Australian dollar'.""" return label if label.endswith("s") else label + "s" def unit_mult_word(unit_mult) -> str: if unit_mult in (None, ""): return "" key = str(unit_mult).strip() if key in UNIT_MULT_WORDS: return UNIT_MULT_WORDS[key] return "units of 10^{}".format(key) def _tokens(text: str) -> List[str]: return [t.lower() for t in re.split(r"[^A-Za-z0-9_@]+", text or "") if t] def _is_mechanism_label(value: Optional[str]) -> bool: if not value: return False cleaned = _squash(str(value)).strip(" .:-—*_[]()").lower() return cleaned in MECHANISM_LABELS def _ends_with_response_format(value: Optional[str]) -> bool: if not value: return False words = _squash(str(value)).strip(" .,;:-—*_()[]").split(" ") return bool(words) and words[-1].lower() in RESPONSE_FORMAT_TOKENS def _named_evidence(evidence: str, prefixes: Tuple[str, ...]) -> bool: """True where the evidence carries one of the prefixes AND names something.""" for prefix in prefixes: if evidence.startswith(prefix) and evidence[len(prefix):].strip(): return True return False def _dedupe(pairs: List[Tuple[Any, Any]]) -> List[Any]: """First-seen order, stable. Aggregation must be reproducible across runs.""" seen = set() out = [] for key, value in pairs: if key in seen: continue seen.add(key) out.append(value) return out # The styles that actually have a renderer behind them. `abs` composes the ABS # clauses, `oecd` adds the derived-by check, `generic` is the skeleton. argparse # and Citation.__init__ both read this so the CLI and the JSON path cannot # advertise different contracts. IMPLEMENTED_STYLES = ("generic", "abs", "oecd") # -------------------------------------------------------------------------- # URN # -------------------------------------------------------------------------- def build_urn(agency: str, artefact_id: str, version: str, artefact_class: str = "dataflow") -> str: cls = URN_CLASSES.get(artefact_class.lower().replace(" ", "")) if cls is None: raise ValueError( f"unknown artefact class {artefact_class!r}; " f"known: {', '.join(sorted(set(URN_CLASSES)))}" ) return (f"urn:sdmx:org.sdmx.infomodel.{cls}=" f"{agency}:{artefact_id}({version})") def parse_triplet(triplet: str): m = TRIPLET_RE.match(triplet.strip()) if not m: raise ValueError( f"{triplet!r} is not a valid AGENCY:ID(VERSION) triplet. " "An unversioned reference is not citable — it silently re-points " "when the provider publishes a new version." ) return m.group(1), m.group(2), m.group(3) # -------------------------------------------------------------------------- # Dates # -------------------------------------------------------------------------- def parse_access_date(raw: str): """The parsed instant, or None where no known format matched.""" raw = (raw or "").strip() if not raw: return None for fmt in ("%Y-%m-%dT%H:%M:%S%z", "%Y-%m-%dT%H:%M:%SZ", "%Y-%m-%dT%H:%M%z", "%Y-%m-%dT%H:%MZ"): try: return datetime.strptime(raw.replace("Z", "+0000") if fmt.endswith("%z") else raw, fmt) except ValueError: continue for fmt in ("%Y-%m-%d", "%d %B %Y", "%d/%m/%Y"): try: return datetime.strptime(raw, fmt).date() except ValueError: continue return None def format_access_date(raw: str) -> str: """Render an access date as 'DD Month YYYY', preserving time where given.""" raw = (raw or "").strip() parsed = parse_access_date(raw) if parsed is None: return raw # pass through unrecognised formats rather than guessing if isinstance(parsed, datetime): # An input carrying a numeric offset must be CONVERTED before it is # labelled UTC. Formatting it as-is recorded the local clock time under # a UTC label, and an offset spanning midnight recorded the wrong date. # # Not every branch of parse_access_date is tz-aware, despite the %z in # most of them: two format strings end in a LITERAL "Z", and strptime # matches format literals case-insensitively, so an input ending in a # lowercase "z" parses naive. astimezone() on a naive datetime assumes # the HOST machine's zone — silently correct on a UTC CI runner and # hours out anywhere else, which is the bug this function exists to # fix. Treat a missing tzinfo as the UTC the trailing Z asserts. if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=timezone.utc) parsed = parsed.astimezone(timezone.utc) return (f"{parsed.day} {parsed.strftime('%B %Y')}, " f"{parsed.strftime('%H:%M')} UTC") return f"{parsed.day} {parsed.strftime('%B %Y')}" # -------------------------------------------------------------------------- # Citation assembly # -------------------------------------------------------------------------- class Citation: """One data unit. The canonical Profile 4 fields live on CitationCollection. Deprecated inputs, kept only so existing callers keep working: title — legacy free-text name. Feeds footnote() and reference_entry() only. It is NEVER promoted to resource_title: translating arbitrary legacy text into a claimed formal title is exactly the failure this contract exists to prevent. accessed — normalised to extracted_at at parse time. Supplying both with different values is a validation error, not a silent winner. """ def __init__(self, **kw): self.agency: Optional[str] = kw.get("agency") self.publisher: Optional[str] = kw.get("publisher") self.data_provider: Optional[str] = kw.get("data_provider") self.title: Optional[str] = kw.get("title") self.reference_period: Optional[str] = kw.get("reference_period") self.medium: Optional[str] = kw.get("medium") self.doi: Optional[str] = kw.get("doi") self.urn: Optional[str] = kw.get("urn") self.query_url: Optional[str] = kw.get("query_url") self.landing_url: Optional[str] = kw.get("landing_url") self.agency_id: Optional[str] = kw.get("agency_id") self.dataflow_id: Optional[str] = kw.get("dataflow_id") self.version: Optional[str] = kw.get("version") self.derived_by: Optional[str] = kw.get("derived_by") self.checksum: Optional[str] = kw.get("checksum") style = kw.get("style") or "generic" # argparse enforces this for the CLI, but `cite --json` (the only way # to render a multi-source block) and direct construction reach here # without passing through it. Validating in one place keeps the two # entry points from diverging again. if style not in IMPLEMENTED_STYLES: raise ValueError( f"unknown style {style!r}; implemented: " f"{', '.join(IMPLEMENTED_STYLES)}" ) self.style: str = style # -- extracted_at is canonical; accessed is the legacy boundary ----- # Strip BEFORE deciding anything. " " is not a vintage, and letting # it through renders "accessed ." with no error. legacy_accessed = (kw.get("accessed") or "").strip() or None extracted_at = (kw.get("extracted_at") or "").strip() or None self.alias_conflict = bool( legacy_accessed and extracted_at and legacy_accessed != extracted_at) self.extracted_at: Optional[str] = extracted_at or legacy_accessed # -- citation-identity fields --------------------------------------- self.resource_title: Optional[str] = kw.get("resource_title") self.resource_kind: Optional[str] = kw.get("resource_kind") self.selected_members: List[dict] = list(kw.get("selected_members") or []) self.selection_labels: List[str] = list(kw.get("selection_labels") or []) self.response_format: Optional[str] = kw.get("response_format") self.licence_label: Optional[str] = kw.get("licence_label") self.licence_url: Optional[str] = kw.get("licence_url") # Strip on the way in. A whitespace-only note is not a note: left # untrimmed it satisfies the "notes are present" gate, then gets # dropped by aggregation, and the mandatory Note field silently # disappears from a block that validated clean. self.notes: List[str] = [ str(n).strip() for n in (kw.get("notes") or []) if str(n).strip()] self.unit_measure: Optional[str] = kw.get("unit_measure") self.unit_mult = kw.get("unit_mult") # Availability states are explicit inputs. They default only in the # direction that BLOCKS publication, never in the direction that # asserts provenance the caller did not claim. self.resource_title_status: str = kw.get("resource_title_status") or ( "verified" if self.resource_title else "unavailable") has_selection = bool(self.selected_members or self.selection_labels) self.selection_basis: str = kw.get("selection_basis") or ( "subset" if has_selection else "whole_resource") self.selection_status: str = kw.get("selection_status") or ( "verified" if has_selection else "unavailable") if not self.urn and self.agency_id and self.dataflow_id: self.urn = build_urn(self.agency_id, self.dataflow_id, self.version or "1.0") # -- deprecated alias: rendering still says "accessed [date]", because # that is reader-facing citation language, not the field name. @property def accessed(self) -> Optional[str]: return self.extracted_at # -- identifier precedence: DOI > URN > query URL > landing page ------ @property def identifier(self) -> Optional[str]: return self.doi or self.urn or self.query_url or self.landing_url @property def triplet(self) -> Optional[str]: if self.agency_id and self.dataflow_id: return f"{self.agency_id}:{self.dataflow_id}({self.version or '1.0'})" return None @property def display_title(self) -> Optional[str]: """Legacy renderings only. Profile 4 requires resource_title.""" return self.title or self.resource_title # -- the display projection ------------------------------------------- def _project_members(self) -> Dict[str, list]: """Walk selected_members once, in DSD order, honouring explicit roles. Returns identity labels, interpretation clauses and the neutral-scope members kept out of the human text. Nothing is dropped from provenance: every member is still in selected_members and still in the query key. """ identity: List[str] = [] interpretation: List[dict] = [] neutral: List[dict] = [] errors: List[str] = [] for i, raw in enumerate(self.selected_members): where = f"selected_members[{i}]" if not isinstance(raw, dict): errors.append(f"{where} is not an object.") continue member = { "dimension_id": (raw.get("dimension_id") or "").strip(), "dimension_label": (raw.get("dimension_label") or "").strip(), "code": (raw.get("code") or "").strip(), "member_label": (raw.get("member_label") or "").strip(), "citation_role": (raw.get("citation_role") or "identity").strip(), "role_evidence": (raw.get("role_evidence") or "").strip(), } role = member["citation_role"] evidence = member["role_evidence"] # All three source-native fields are mandatory. A member with a # label but no dimension and no code names nothing checkable — it # is a sentence, not a selection. if not member["member_label"]: errors.append( f"{where} has no member_label. A selected member without " f"the provider's label cannot be cited by name.") if not member["dimension_id"]: errors.append(f"{where} has no dimension_id.") if not member["code"]: errors.append(f"{where} has no code.") if role not in CITATION_ROLES: errors.append( f"{where} has citation_role {role!r}; expected one of " f"{', '.join(CITATION_ROLES)}.") continue if not evidence: errors.append( f"{where} has no role_evidence. Record the bound concept, " f"the provider aggregate evidence, or the sentinel " f"{DEFAULT_IDENTITY!r}.") continue if evidence == DEFAULT_IDENTITY and role != "identity": errors.append( f"{where} claims role {role!r} on {DEFAULT_IDENTITY!r} " f"evidence. That sentinel only justifies identity.") continue # A prefix alone is not evidence — "concept:" with nothing after it # names no concept. Require something after the colon. if role == "interpretation" and not _named_evidence( evidence, (INTERPRETATION_EVIDENCE,)): errors.append( f"{where} is interpretation but role_evidence does not " f"name a bound concept ('{INTERPRETATION_EVIDENCE}').") continue if role == "neutral_scope" and not _named_evidence( evidence, NEUTRAL_SCOPE_EVIDENCE): errors.append( f"{where} is neutral_scope but role_evidence does not name " f"provider codelist or aggregate evidence " f"({' / '.join(NEUTRAL_SCOPE_EVIDENCE)}). An English " f"label reading 'Total' is not evidence.") continue if role == "identity": identity.append(member["member_label"]) elif role == "interpretation": interpretation.append(member) else: neutral.append(member) return { "identity_labels": identity, "interpretation": interpretation, "neutral_scope": neutral, "clauses": self._interpretation_clauses(interpretation), "errors": errors, } def _interpretation_clauses(self, members: List[dict]) -> List[str]: if self.style == "abs": return self._abs_clauses(members) return self._generic_clauses(members) def _generic_clauses(self, members: List[dict]) -> List[str]: clauses = [self._labelled_clause(m) for m in members] # A unit DIMENSION already rendered its own clause. Adding the # unit_measure attribute on top says the same thing twice in different # words — which the note dedupe cannot catch, because the two strings # differ. The multiplier still needs somewhere to go. has_unit_member = any(_concept_of(m) in UNIT_CONCEPTS for m in members) unit = self._unit_clause(skip_measure=has_unit_member) if unit: clauses.append(unit) return clauses def _abs_clauses(self, members: List[dict]) -> List[str]: """ABS Profile 4 composition, scoped to the canonical block. Composes the verified adjustment, frequency, unit and multiplier into the two clauses the ABS house style uses. Anything it does not recognise falls through to the generic labelled form rather than being dropped — an unrecognised interpretation dimension is still meaning. """ adjustment = frequency = unit_label = None rest: List[dict] = [] for member in members: concept = _concept_of(member) if concept in ADJUSTMENT_CONCEPTS and adjustment is None: adjustment = member["member_label"] elif concept in FREQ_CONCEPTS and frequency is None: frequency = member["member_label"] elif concept in UNIT_CONCEPTS and unit_label is None: unit_label = member["member_label"] else: rest.append(member) clauses: List[str] = [] observation_bits: List[str] = [] if adjustment: observation_bits.append(adjustment) if frequency: observation_bits.append( frequency.lower() if observation_bits else frequency) if observation_bits: clauses.append(" ".join(observation_bits) + " observations") unit_bits: List[str] = [] if unit_label: unit_bits.append(_pluralise(unit_label)) elif self.unit_measure: # No unit dimension among the members, so fall back to the # UNIT_MEASURE attribute — verbatim, because that is typically a # code ("AUD") and pluralising a code produces "AUDs". unit_bits.append(self.unit_measure) word = unit_mult_word(self.unit_mult) if word: unit_bits.append(word) if unit_bits: clauses.append(", ".join(unit_bits)) clauses.extend(self._labelled_clause(m) for m in rest) return clauses @staticmethod def _labelled_clause(member: dict) -> str: label = member["dimension_label"] or member["dimension_id"] or "Dimension" return f"{label}: {member['member_label']}" def _unit_clause(self, skip_measure: bool = False) -> Optional[str]: word = unit_mult_word(self.unit_mult) measure = None if skip_measure else self.unit_measure if measure and word: return f"Unit: {measure}, {word}" if measure: return f"Unit: {measure}" if word: return f"Unit multiplier: {word}" return None def _codes_absent_from_query(self) -> List[str]: """Codes the query URL does not appear to contain. Both sides are tokenised the same way, so a code carrying a separator (`EU27-2020`, `OECD.SDD`) is compared piece by piece rather than being reported missing because the URL split it. Dotted SDMX keys, `+`-joined multi-member selections, `?key=` forms and wildcards all tokenise cleanly under this rule. The URL is percent-decoded first, and the comparison is case-insensitive: Eurostat documents its component filters (`c[geo]=eu27_2020`) as case-insensitive, so a case-sensitive match would warn on a valid query. """ if not (self.query_url and self.selected_members): return [] tokens = set(_tokens(unquote(self.query_url))) missing = [] for raw in self.selected_members: if not isinstance(raw, dict): continue code = (raw.get("code") or "").strip() parts = _tokens(code) if parts and not all(p in tokens for p in parts): missing.append(code) return missing def projected_selection_labels(self) -> List[str]: if self.selected_members: return self._project_members()["identity_labels"] return [s for s in self.selection_labels if str(s).strip()] def projected_note_clauses(self) -> List[str]: if self.selected_members: return self._project_members()["clauses"] unit = self._unit_clause() return [unit] if unit else [] def note_clauses(self) -> List[str]: """Caller-supplied notes win, but only after validation checks that the projected clauses survive inside them, in order.""" if self.notes: return [str(n) for n in self.notes] return self.projected_note_clauses() def selection_label(self) -> Optional[str]: labels = self.projected_selection_labels() return SELECTION_JOINER.join(labels) if labels else None # -- validation ------------------------------------------------------- def validate(self) -> Dict[str, List[str]]: errors, warnings = [], [] if not self.agency: errors.append("No agency. The maintaining agency is the author " "and cannot be inferred.") if not self.display_title: errors.append("No title.") if not self.extracted_at: errors.append( "No access/extraction date. Agency databases are continuously " "revised, so a citation without one is not reproducible.") if not self.identifier: errors.append( "No identifier. Need at least one of DOI, URN or query URL — " "a landing page alone does not identify the numbers cited.") if self.alias_conflict: errors.append( "accessed and extracted_at disagree. accessed is a deprecated " "alias for extracted_at; supply one, not two vintages.") errors.extend(self.profile4_errors()) if not self.reference_period: warnings.append( "No reference period. This is the period the data describes " "and is distinct from the publication year and from the " "extraction date.") if self.version is None and self.dataflow_id: warnings.append( "No artefact version; defaulting to 1.0. An unversioned " "reference resolves to whatever is in production today.") if self.query_url and not self.doi and not self.urn: warnings.append( "Query URL is the only identifier. Add a URN (or DOI if the " "provider mints one) — a URL identifies a location, not an " "identity.") if self.identifier == self.landing_url: warnings.append( "The only identifier is a landing page. A landing page renders " "a current view, not the view cited — add a DOI, URN or the " "query/table reference that resolves to these observations.") if self.landing_url and not self.query_url and self.dataflow_id: warnings.append( "Landing page given but no data query URL. The query is the " "reproducible artifact; the landing page renders a current " "view, not the view cited.") if not self.checksum and self.query_url: warnings.append( "No result-set checksum. Optional, but it is what lets a " "reader verify they got the same data back (RDA R6).") if self.extracted_at and parse_access_date(self.extracted_at) is None: warnings.append( f"extracted_at {self.extracted_at!r} matched no known date " f"format, so it is rendered verbatim. A vintage a reader " f"cannot parse is a vintage they cannot use.") if self.derived_by and self.style == "oecd": warnings.append( "OECD style forbids 'Source: Authors' alone — name the " "underlying dataset as the data source as well as flagging " "the derivation.") if self.selection_labels and not self.selected_members: warnings.append( "selection_labels supplied without selected_members. Labels " "alone cannot prove a member's role or a subset's " "applicability — supply selected_members unless this is a " "provider-authored formal series title.") # Cheap sanity binding between the labelled members and the query they # claim to describe. A warning, not an error: key syntax varies across # providers (path key, ?key=, wildcards, +-joined members), so a strict # parse would reject valid citations. It still catches the case that # matters — labels pasted onto an unrelated query. missing = self._codes_absent_from_query() if missing: warnings.append( "selected_members codes {} do not appear in query_url. Either " "the labels describe a different query, or the provider's key " "syntax hides them — check before publishing.".format( ", ".join(repr(c) for c in missing))) # A warning, not an error: a provider could in principle publish a # title ending in a format token, and refusing to render a real title # is worse than flagging it. for field, value in (("resource_title", self.resource_title), ("selection label", self.selection_label())): if _ends_with_response_format(value): warnings.append( f"{field} {value!r} ends in a response format. The format " f"is a transport fact and belongs after the link, in " f"parentheses — not inside a name.") return {"errors": errors, "warnings": warnings} def profile4_errors(self) -> List[str]: """Publication-blocking checks for the canonical source block. A unit failing these is still a valid unit — it carries its codes, its triplet and its query. What it is not is publishable as a NAMED citation, and the correct response is a codes-only citation plus a report of the unresolved metadata. """ errors: List[str] = [] projection = self._project_members() if self.selected_members else None if projection: errors.extend("Profile 4: " + e for e in projection["errors"]) if self.resource_title_status not in TITLE_STATUSES: errors.append( f"Profile 4: resource_title_status is " f"{self.resource_title_status!r}; expected " f"{' or '.join(TITLE_STATUSES)}.") elif self.resource_title_status != "verified": errors.append( "Profile 4: the formal title is unavailable. Keep the " "identifier and dimension codes, report that the official " "title must be resolved, and do not publish a named block.") elif not (self.resource_title or "").strip(): errors.append( "Profile 4: resource_title_status is verified but no " "resource_title was supplied.") # resource_kind records what supplied the formal TITLE, so demanding it # when the title is unavailable asks which kind of artefact supplied a # name that does not exist. That state is already blocked; adding a # second, meaningless error just buries the real one. A kind that was # supplied and is malformed is still rejected either way. if self.resource_kind is not None: if self.resource_kind not in RESOURCE_KINDS: errors.append( f"Profile 4: resource_kind is {self.resource_kind!r}; " f"expected {' or '.join(RESOURCE_KINDS)}.") elif self.resource_title_status == "verified": errors.append( f"Profile 4: no resource_kind; expected " f"{' or '.join(RESOURCE_KINDS)} — it records what supplied the " f"formal title.") if self.selection_basis not in SELECTION_BASES: errors.append( f"Profile 4: selection_basis is {self.selection_basis!r}; " f"expected {' or '.join(SELECTION_BASES)}.") if self.selection_status not in SELECTION_STATUSES: errors.append( f"Profile 4: selection_status is {self.selection_status!r}; " f"expected {', '.join(SELECTION_STATUSES)}.") elif self.selection_status == "unavailable": errors.append( "Profile 4: the selection is unavailable. A query naming a " "subset whose labels will not resolve is blocked, not " "rendered with a substitute label.") elif self.selection_status == "not_applicable": if self.selection_basis != "whole_resource": errors.append( "Profile 4: selection_status=not_applicable requires " "selection_basis=whole_resource. A subset always has a " "selection to name.") if self.projected_selection_labels(): errors.append( "Profile 4: selection_status=not_applicable but selection " "labels were supplied. Say which one is true.") else: # verified labels = self.projected_selection_labels() if not labels: errors.append( "Profile 4: selection_status=verified but no selection " "label survived the projection.") elif not self.selected_members and len(labels) != 1: # Applies whatever the basis says. Labels alone cannot prove a # member's role, so the only labels-only selection that can be # trusted is one the provider itself titled — and that is one # label, not a list someone assembled by hand. errors.append( "Profile 4: a verified selection needs selected_members. " "A labels-only selection is accepted only as a single " "provider-authored formal series title.") for i, label in enumerate(self.selection_labels): if not isinstance(label, str) or not label.strip(): errors.append( f"Profile 4: selection_labels[{i}] is not a non-empty " f"string.") # The link carries the title or the selection, so a mechanism label # can only get in through one of those two fields. Block it there. for field, value in (("resource_title", self.resource_title), ("title", self.title)): if _is_mechanism_label(value): errors.append( f"Profile 4: {field} {value!r} names the retrieval " f"mechanism, not the data. Use the provider's title.") for i, label in enumerate(self.projected_selection_labels()): if _is_mechanism_label(label): errors.append( f"Profile 4: selection label {label!r} names the retrieval " f"mechanism, not the data.") if projection and self.selection_labels: projected = projection["identity_labels"] if list(self.selection_labels) != projected: errors.append( "Profile 4: selection_labels do not match the projection " "from selected_members. Expected " f"{projected!r} in that order; the roles decide the " "labels, not the caller.") if self.notes: supplied = [_squash(str(n)) for n in self.notes] required = [_squash(c) for c in self.projected_note_clauses()] cursor = 0 for clause in required: try: cursor = supplied.index(clause, cursor) + 1 except ValueError: errors.append( f"Profile 4: note clause {clause!r} is required by the " f"projection but is absent or out of order in notes.") break if not self.note_clauses(): errors.append( "Profile 4: no note clauses. Interpretation metadata — " "adjustment, frequency, unit, multiplier — is what makes the " "number readable; state 'none' explicitly if there truly is " "none.") if not self.extracted_at: errors.append( "Profile 4: no extracted_at. Whitespace is not a vintage.") # Whitespace is not a licence. Strip before deciding whether a field # was supplied, or " " renders as an empty link label. has_label = bool((self.licence_label or "").strip()) has_url = bool((self.licence_url or "").strip()) if has_label != has_url: errors.append( "Profile 4: licence_label and licence_url must be supplied " "together. Half a licence is not a confirmed licence.") if not has_label: errors.append( "Profile 4: no confirmed licence. Confirm it per dataset — " "never from the agency name or an assumed portal default.") if self.licence_url and not is_http_url(self.licence_url): errors.append( f"Profile 4: licence_url {self.licence_url!r} is not an " f"absolute http(s) URL.") if self.query_url and not is_http_url(self.query_url): errors.append( f"Profile 4: query_url {self.query_url!r} is not an absolute " f"http(s) URL.") if not self.query_url: if self.projected_selection_labels(): errors.append( "Profile 4: a selection without a query URL. The link is " "what makes the selection reproducible.") if self.response_format: errors.append( "Profile 4: a response format without a query URL. The " "format describes a response nothing here can fetch.") return errors # -- canonical renderings --------------------------------------------- def source_entry(self) -> str: """One source entry: title, identifier, linked selection, format, date. No field label — the collection owns the singular `Source:` label so that N entries can never become N labels. """ head = self.agency or "[agency]" if self.reference_period: head += f" ({self.reference_period})" title = self.resource_title or self.display_title or "[title]" title_md = f"*{title}*" selection = self.selection_label() if not selection and self.query_url: # Nothing else may carry the link, and a mechanism label is never # an option — so the official title carries it. title_md = f"[{title_md}]({self.query_url})" entry = f"{head}, {title_md}" bracket = [b for b in (self.medium, self.triplet) if b] if bracket: entry += " [" + ", ".join(bracket) + "]" for extra in _dedupe([(v, v) for v in (self.publisher, self.data_provider) if v and v != self.agency]): entry += f", {extra}" if self.doi: entry += f", {self.doi}" if selection: link = (f"[{selection}]({self.query_url})" if self.query_url else selection) entry += f". Selection: {link}" if self.response_format: entry += f" ({self.response_format})" if self.extracted_at: entry += f", accessed {format_access_date(self.extracted_at)}" return entry def licence_entry(self) -> Optional[str]: """One normalised licence link, without the field label.""" if self.licence_label and self.licence_url: return f"[{self.licence_label}]({self.licence_url})" return self.licence_label or None def source_key(self) -> tuple: """Agency/dataflow/version + normalised query identity + vintage. extracted_at is in the key deliberately: two extractions of one query at different vintages are two citable things. The last two components are a strictly hierarchical fallback, used only where the components above them cannot discriminate: query identity -> nothing further needed else triplet -> nothing further needed else identifier -> DOI / URN / landing page else title -> the last resort Without it, a DOI-identified table with no triplet and no query URL keys on agency plus date alone, so two different datasets published by one agency on one day collapse and the second vanishes from the block. Adding the title unconditionally would be the opposite bug: two citations of one versioned artefact whose titles are represented differently would stop deduplicating. """ query = normalize_query_url(self.query_url) discriminated = bool(query) or bool(self.triplet) fallback_id = "" if discriminated else normalize_query_url( self.doi or self.urn or self.landing_url or "") fallback_title = "" if (discriminated or fallback_id) else _squash( self.resource_title or self.display_title or "") return ( self.agency_id or self.agency or "", self.dataflow_id or "", self.version or "", query, (self.extracted_at or "").strip(), fallback_id, fallback_title, ) def licence_key(self) -> tuple: # The label is free text, so it needs its own normalisation rule or the # key is not a key: "CC BY 4.0 (ABS terms)" and "CC BY 4.0 (ABS terms)" # are the same licence and must collapse. return (normalize_query_url(self.licence_url) or (self.licence_url or ""), _squash(self.licence_label or "")) def data_link(self) -> Optional[Dict[str, Optional[str]]]: if not self.query_url: return None return { "label": self.selection_label() or self.resource_title or self.display_title, "url": self.query_url, "response_format": self.response_format, } def licence(self) -> Optional[Dict[str, Optional[str]]]: if not (self.licence_label or self.licence_url): return None return {"label": self.licence_label, "url": self.licence_url} # -- legacy renderings ------------------------------------------------- def _core(self) -> str: bits: List[str] = [] if self.agency: bits.append(self.agency) if self.reference_period: bits.append(f"({self.reference_period})") head = " ".join(bits) parts = [p for p in [head, self.display_title] if p] s = " ".join(parts) # Medium and triplet are different facts and both belong in the # bracketed slot: the medium says how it was accessed, the triplet # says which versioned artefact was accessed. bracket = [b for b in (self.medium, self.triplet) if b and b not in (s or "")] if bracket: s += " [" + ", ".join(bracket) + "]" # Publisher and data provider are separate roles but often one body; # deduplicate rather than warning about it. for extra in _dedupe([(v, v) for v in (self.publisher, self.data_provider) if v and v != self.agency]): s += f", {extra}" if self.identifier: s += f", {self.identifier}" if self.query_url and self.identifier != self.query_url: s += f", {self.query_url}" if self.checksum: s += f", result checksum: {self.checksum}" if self.extracted_at: s += f", accessed {format_access_date(self.extracted_at)}" return s def footnote(self) -> str: s = self._core() if self.derived_by: s = f"Calculated by {self.derived_by} from {s}" return s.rstrip(".") + "." def table_source(self) -> str: """Deprecated: use CitationCollection.source_block().""" body = self._core().rstrip(".") if self.derived_by: return (f"Source: calculated by {self.derived_by} from {body}. " f"{self.agency or 'The source agency'} is not responsible " f"for the calculation.") return f"Source: {body}." def reference_entry(self) -> str: """DataCite-preferred rendering: Creator (Year): Title. Version. Publisher. (type). Identifier""" year = self.reference_period or "" parts = [f"{self.agency or '[agency]'} ({year}):" if year else f"{self.agency or '[agency]'}:", f"{self.display_title or '[title]'}."] if self.version: parts.append(f"Version {self.version}.") if self.publisher and self.publisher != self.agency: parts.append(f"{self.publisher}.") if self.medium: parts.append(f"({self.medium}).") if self.identifier: parts.append(self.identifier) s = " ".join(parts) if self.extracted_at: s += f" Accessed {format_access_date(self.extracted_at)}." return s def compact(self) -> Optional[str]: """Deprecated: a one-item collection rendered through profile4(). The old lossy agency-plus-triplet form is gone on purpose — it dropped the selection, the licence and the link, which is most of what a dashboard footer has to carry. """ return CitationCollection([self]).profile4() def render_all(self) -> Dict[str, Optional[str]]: return { "footnote": self.footnote(), "table_source": self.table_source(), "reference_entry": self.reference_entry(), "compact": self.compact(), } def _concept_of(member: dict) -> str: """The bound concept named in role_evidence, upper-cased, or ''.""" evidence = member.get("role_evidence") or "" if not evidence.startswith(INTERPRETATION_EVIDENCE): return "" return evidence[len(INTERPRETATION_EVIDENCE):].strip().upper() # -------------------------------------------------------------------------- # Multi-source aggregation. One code path owns the singular field labels, so # a one-item block and a three-item block cannot drift apart. # -------------------------------------------------------------------------- class CitationCollection: def __init__(self, citations: List[Citation]): self.citations: List[Citation] = list(citations) def validate(self) -> Dict[str, List[str]]: errors: List[str] = [] warnings: List[str] = [] many = len(self.citations) > 1 for i, citation in enumerate(self.citations): checks = citation.validate() prefix = f"[{i}] " if many else "" errors.extend(prefix + e for e in checks["errors"]) warnings.extend(prefix + w for w in checks["warnings"]) if not self.citations: errors.append("No citations supplied.") return {"errors": errors, "warnings": warnings} def blocked(self) -> bool: return bool(self.validate()["errors"]) # -- deduplicated entries --------------------------------------------- def source_entries(self) -> List[str]: return _dedupe([(c.source_key(), c.source_entry()) for c in self.citations]) def licence_entries(self) -> List[str]: pairs = [(c.licence_key(), c.licence_entry()) for c in self.citations if c.licence_entry()] return _dedupe(pairs) def note_clauses(self) -> List[str]: pairs = [] for citation in self.citations: for clause in citation.note_clauses(): text = str(clause).strip() if text: pairs.append((_squash(text), text)) return _dedupe(pairs) # -- the three singular fields ---------------------------------------- def source_line(self) -> str: return "Source: " + ENTRY_JOINER.join(self.source_entries()) + "." def licence_line(self) -> Optional[str]: entries = self.licence_entries() if not entries: return None return "Licence: " + ENTRY_JOINER.join(entries) + "." def note_line(self) -> Optional[str]: clauses = self.note_clauses() if not clauses: return None return "Note: " + ENTRY_JOINER.join(clauses) + "." def profile4_fields(self) -> Optional[Dict[str, Optional[str]]]: if self.blocked(): return None return { "source": self.source_line(), "licence": self.licence_line(), "note": self.note_line(), } def source_block(self) -> Optional[str]: fields = self.profile4_fields() if fields is None: return None return "\n".join(v for v in (fields["source"], fields["licence"], fields["note"]) if v) def profile4(self) -> Optional[str]: return self.source_block() # -------------------------------------------------------------------------- # CLI # -------------------------------------------------------------------------- def load_citations(payload) -> List[Citation]: """Accept the legacy single-citation object or {"citations": [...]}. JSON is the only multi-source input; direct CLI flags always build a one-item collection, because a flag namespace cannot express which member belongs to which source. """ if isinstance(payload, dict) and "citations" in payload: items = payload["citations"] if not isinstance(items, list): raise ValueError('"citations" must be a list of citation objects') return [Citation(**item) for item in items] return [Citation(**payload)] def cmd_cite(args) -> int: if args.json: with open(args.json) as fh: payload = json.load(fh) citations = load_citations(payload) else: members = [] for raw in args.selected_member_json or []: members.append(json.loads(raw)) citations = [Citation( agency=args.agency, publisher=args.publisher, data_provider=args.data_provider, title=args.title, reference_period=args.reference_period, medium=args.medium, doi=args.doi, urn=args.urn, query_url=args.query_url, landing_url=args.landing_url, accessed=args.accessed, extracted_at=args.extracted_at, agency_id=args.agency_id, dataflow_id=args.dataflow_id, version=args.version, derived_by=args.derived_by, checksum=args.checksum, style=args.style, resource_title=args.resource_title, resource_kind=args.resource_kind, resource_title_status=args.resource_title_status, selected_members=members, selection_labels=args.selection_label or [], selection_basis=args.selection_basis, selection_status=args.selection_status, response_format=args.response_format, licence_label=args.licence_label, licence_url=args.licence_url, notes=args.note or [], unit_measure=args.unit_measure, unit_mult=args.unit_mult, )] collection = CitationCollection(citations) checks = collection.validate() block = collection.source_block() out: Dict[str, Any] = {} if block is not None: out["source_block"] = block out["profile4_fields"] = collection.profile4_fields() out["citations"] = [ { "renderings": c.render_all(), "urn": c.urn, "triplet": c.triplet, "identifier": c.identifier, "data_link": c.data_link(), "licence": c.licence(), "selection_labels": c.projected_selection_labels(), "notes": c.note_clauses(), "neutral_scope": c._project_members()["neutral_scope"] if c.selected_members else [], } for c in citations ] out.update(checks) if len(citations) == 1: only = citations[0] out["renderings"] = only.render_all() out["data_link"] = only.data_link() out["licence"] = only.licence() if only.urn: out["urn"] = only.urn if only.triplet: out["triplet"] = only.triplet if args.format == "json": print(json.dumps(out, indent=2, ensure_ascii=False)) else: # The canonical block comes first: it is the artifact, the bibliography # renderings are the specialist forms. if block is not None: print(f"\n[source_block]\n{block}") else: print("\n[source_block]\nblocked — see errors below. Keep the " "identifiers and dimension codes, report the unresolved " "metadata, and do not publish a named block.") for i, citation in enumerate(citations): tag = f" {i}" if len(citations) > 1 else "" for name, text in citation.render_all().items(): if name == "compact" or text is None: continue print(f"\n[{name}{tag}]\n{text}") if citation.urn: print(f"\n[urn{tag}]\n{citation.urn}") if checks["errors"]: print("\nERRORS — do not publish until resolved:") for e in checks["errors"]: print(f" - {e}") if checks["warnings"]: print("\nWarnings:") for w in checks["warnings"]: print(f" - {w}") if not checks["errors"] and not checks["warnings"]: print("\nProvenance complete.") return 1 if checks["errors"] else 0 def cmd_flag(args) -> int: src = args.source.lower() if args.source else None # A supplied codelist is authoritative for whatever it decodes: it is the # list this dataflow's DSD actually points at. codes: Optional[Dict[str, str]] = None provenance = "" if args.codelist: try: with open(args.codelist, encoding="utf-8") as fh: codes, provenance = load_codelist(json.load(fh)) except (OSError, ValueError, json.JSONDecodeError) as exc: print(f"could not read codelist {args.codelist!r}: {exc}", file=sys.stderr) return 2 if src is not None and src not in FLAGS and src not in SDMX_ATTRIBUTES: known = sorted(list(FLAGS) + list(SDMX_ATTRIBUTES)) print(f"Unknown source {args.source!r}. Known: {', '.join(known)}", file=sys.stderr) return 2 # An SDMX attribute with no codelist supplied is the case this command # refuses to guess at. It used to answer from a table shipped in this file, # which is how a cross-domain default came to be published as though it were # the provider's own semantics. if codes is None: if src in SDMX_ATTRIBUTES: attr = SDMX_ATTRIBUTES[src] print( f"{args.code!r} is UNRESOLVED." if args.code else f"{attr} is UNRESOLVED.", file=sys.stderr) print( f"\n{attr} is a DSD-declared attribute: the data structure " f"names the codelist that decodes it, and providers often " f"point at their own rather than the SDMX cross-domain list. " f"Nothing is hardcoded here, deliberately — a shipped table " f"answers confidently for a dataflow it has never seen.\n\n" f"Resolve it, then pass it back:\n" f" 1. inspect(agency_id, dataflow_id) and read the " f"{attr} attribute's codelist reference\n" f" 2. retrieve that codelist\n" f" 3. re-run with --codelist \n\n" f"If you cannot resolve it, report the raw code and the " f"codelist it belongs to, and publish no meaning for it.", file=sys.stderr) return 1 if src is None: print("Give --source (a publication convention) or --codelist " "(a resolved SDMX code list).", file=sys.stderr) return 2 table = codes if codes is not None else FLAGS[src] label = provenance if codes is not None else src if args.list: for code, meaning in table.items(): print(f"{code:<8} {meaning}") return 0 code = args.code if code is None: print("Give --code or --list", file=sys.stderr) return 2 meaning = table.get(code) if meaning is None: elsewhere = [s2 for s2 in FLAGS if code in FLAGS[s2] and s2 != src] hint = (f" It IS in: {', '.join(elsewhere)} — but those are publication " f"conventions, not this code list; never map between them " f"unchecked." if elsewhere else "") print(f"{code!r} is not in {label}. " f"Do not guess — check the source's own documentation and the " f"extraction vintage.{hint}", file=sys.stderr) return 1 print(f"{label}: {code} = {meaning}") # A resolved SDMX code that also exists as an agency publication symbol is # worth flagging: the same character carries unrelated meanings across the # two, and only the code list above applies to this observation. others = [s2 for s2 in FLAGS if s2 != src and code in FLAGS[s2]] if others: print("\nSame character, different meaning elsewhere " "(never map between systems unchecked):") for s2 in others: print(f" {s2}: {code} = {FLAGS[s2][code]}") if src == "eurostat": print("\nVintage note: from 27 January 2025 Eurostat splits flags into " "Obs_status and Conf_status code lists, introduces N and P, and " "recodes three earlier flags. Check the extraction date.") if src == "abs" and code in ("*", "**"): print("\nVintage note: older ABS publications use * for high standard " "errors and ** for RSE > 50%, which differs from the current " "banding. Check the publication vintage.") return 0 def cmd_urn(args) -> int: if args.triplet: agency, aid, version = parse_triplet(args.triplet) else: if not (args.agency_id and args.id): print("Give --triplet, or --agency-id and --id", file=sys.stderr) return 2 agency, aid, version = args.agency_id, args.id, args.version or "1.0" print(build_urn(agency, aid, version, args.artefact_class)) return 0 def main(argv=None) -> int: p = argparse.ArgumentParser( description="Citation and provenance tooling for official statistics.") sub = p.add_subparsers(dest="cmd", required=True) c = sub.add_parser( "cite", help="assemble and validate a citation", description="Render the canonical Source / Licence / Note block. " "The official title, the formal selection, the query URL " "and the response format are four different things and " "have four different flags.") c.add_argument("--json", help="read all fields from a JSON file instead — " "one citation object, or {\"citations\": [...]} " "for a multi-source block") c.add_argument("--agency", help="maintaining agency / author") c.add_argument("--publisher", help="publisher if different from agency") c.add_argument("--data-provider", help="body disseminating the data") c.add_argument("--title", help="DEPRECATED legacy free-text name. Feeds the " "footnote and reference-list forms only and " "is never treated as a formal title") c.add_argument("--reference-period", help="period the DATA describes, e.g. 2021-Q3 or 2015-2024") c.add_argument("--medium", help="e.g. 'Data Explorer', 'SDMX dataflow'") c.add_argument("--doi") c.add_argument("--urn", help="built from the triplet if omitted") c.add_argument("--query-url", help="REST data query — the technical reproducible artifact. " "Its LINK TEXT is the title or selection, never a " "mechanism label") c.add_argument("--landing-url") c.add_argument("--accessed", help="DEPRECATED alias for --extracted-at; supplying both " "with different values is an error") c.add_argument("--extracted-at", help="extraction date/timestamp — the canonical vintage field") c.add_argument("--agency-id", help="SDMX agencyID, e.g. ESTAT") c.add_argument("--dataflow-id") c.add_argument("--version", help="artefact version; defaults to 1.0") c.add_argument("--derived-by", help="author of a calculated figure") c.add_argument("--checksum", help="result-set checksum (RDA R6)") c.add_argument("--resource-title", help="the PROVIDER'S title for the whole resource, verbatim") c.add_argument("--resource-kind", choices=list(RESOURCE_KINDS), help="what supplied the formal title") c.add_argument("--resource-title-status", choices=list(TITLE_STATUSES), help="defaults to verified when a title is given, else " "unavailable") c.add_argument("--selected-member-json", action="append", help="repeatable; one JSON object per selected dimension, in " "DSD order, with dimension_id, dimension_label, code, " "member_label, citation_role and role_evidence") c.add_argument("--selection-label", action="append", help="DEPRECATED except for a provider-authored formal " "series title — labels alone cannot prove a member's " "role or a subset's applicability") c.add_argument("--selection-basis", choices=list(SELECTION_BASES)) c.add_argument("--selection-status", choices=list(SELECTION_STATUSES)) c.add_argument("--response-format", help="CSV, SDMX-CSV, JSON — the transport representation, " "rendered outside the link and outside every name") c.add_argument("--licence-label", help="confirmed per-dataset licence name") c.add_argument("--licence-url", help="canonical terms URL for that licence") c.add_argument("--note", action="append", help="repeatable; order is preserved") c.add_argument("--unit-measure", help="UNIT_MEASURE, for the Note clause") c.add_argument("--unit-mult", help="UNIT_MULT power of ten, e.g. 6") c.add_argument("--style", default="generic", choices=list(IMPLEMENTED_STYLES), help="selects the Profile 4 note composition. Only these " "three are implemented: 'abs' composes the ABS " "clauses, 'oecd' adds the derived-by check, 'generic' " "is the default skeleton. The legacy bibliography " "renderings are style-independent") c.add_argument("--format", default="text", choices=["text", "json"], help="output encoding — unrelated to --response-format") c.set_defaults(func=cmd_cite) f = sub.add_parser("flag", help="resolve a data-quality flag") f.add_argument("--source", choices=sorted(list(FLAGS.keys()) + list(SDMX_ATTRIBUTES)), help="publication convention to resolve against. The sdmx " "/ sdmx_conf names refer to DSD-declared attributes " "and need --codelist; they are not carried here") f.add_argument("--codelist", help="JSON code list the DSD points at, from inspect(). " "Authoritative when given — accepts either " "{id, agency_id, version, codes:{...}} or an SDMX-JSON " "structure message") f.add_argument("--code") f.add_argument("--list", action="store_true", help="list the whole set") f.set_defaults(func=cmd_flag) u = sub.add_parser("urn", help="build or check an SDMX URN") u.add_argument("--triplet", help="AGENCY:ID(VERSION)") u.add_argument("--agency-id") u.add_argument("--id") u.add_argument("--version") u.add_argument("--artefact-class", default="dataflow", choices=sorted(set(URN_CLASSES.keys()))) u.set_defaults(func=cmd_urn) args = p.parse_args(argv) try: return args.func(args) except ValueError as exc: print(f"error: {exc}", file=sys.stderr) return 2 if __name__ == "__main__": raise SystemExit(main())