Dataset Schema Framework

const ISO_DATE_TIME = /^(\d{4})-(\d{2})-(\d{2})(?:T(\d{2}):(\d{2})(?::(\d{2}))?(Z|[+-]\d{2}:\d{2})?)?$/;

function isAbsoluteUrl(value) {
  return typeof value === "string" && /^https?:\/\/[^\s/?#]+\.[^\s/?#]+(?:[/?#]\S*)?$/i.test(value);
}

// Returns the ISO 8601 string unchanged (timezone offset kept), or null if unparseable.
function toIsoDateTime(value) {
  if (typeof value !== "string") return null;
  const text = value.trim();
  const m = text.match(ISO_DATE_TIME);
  if (!m) return null;
  const [, y, mo, d, h = "00", mi = "00", s = "00"] = m;
  const date = new Date(Date.UTC(+y, +mo - 1, +d));
  if (date.getUTCMonth() !== +mo - 1 || date.getUTCDate() !== +d) return null;
  if (+h > 23 || +mi > 59 || +s > 59) return null;
  return text;
}

function toIsoDate(value) {
  const iso = toIsoDateTime(value);
  return iso ? iso.slice(0, 10) : null;
}

// Drops null, undefined, "", [] and {} recursively.
function compact(value) {
  if (Array.isArray(value)) {
    const items = value.map(compact).filter((v) => v !== undefined);
    return items.length ? items : undefined;
  }
  if (value && typeof value === "object") {
    const out = {};
    for (const [key, v] of Object.entries(value)) {
      const c = compact(v);
      if (c !== undefined) out[key] = c;
    }
    return Object.keys(out).length ? out : undefined;
  }
  return value === null || value === undefined || value === "" ? undefined : value;
}

function passesPageGates(source) {
  if (source.indexable === false) return false;
  if (source.canonicalUrl && source.url && source.canonicalUrl !== source.url) return false;
  return source.contentVisible !== false;
}


const ORCID = /^https:\/\/orcid\.org\/\d{4}-\d{4}-\d{4}-\d{3}[\dX]$/;
const ROR = /^https:\/\/ror\.org\/0[a-hj-km-np-tv-z0-9]{6}\d{2}$/;
const DOI = /^10\.\d{4,9}\/\S+$/;
const PARTIAL_DATE = /^(\d{4})(?:-(\d{2})(?:-(\d{2}))?)?$/;

function textLength(value) {
  return typeof value === "string" ? value.trim().length : 0;
}

// "2016", "2016-05", "2016-05-01" or a full ISO 8601 date time.
function toTemporalPoint(value) {
  const m = typeof value === "string" ? value.match(PARTIAL_DATE) : null;
  if (m && m[3]) return toIsoDate(value);
  if (m) return !m[2] || (+m[2] >= 1 && +m[2] <= 12) ? value : null;
  return value && value.includes("T") ? toIsoDateTime(value) : null;
}

// ISO 8601 date or interval: "2008", "1950-01-01/2013-12-18", or "2013-12-19/.." for an open end.
function toTemporalCoverage(value) {
  if (typeof value !== "string") return null;
  const [start, end, extra] = value.trim().split("/");
  if (extra !== undefined || !toTemporalPoint(start)) return null;
  if (end === undefined) return start;
  if (end === "..") return `${start}/..`;
  if (!toTemporalPoint(end)) return null;
  const first = start.length === 4 ? `${start}-01-01` : start.length === 7 ? `${start}-01` : start.slice(0, 10);
  const last = end.length === 4 ? `${end}-12-31` : end.length === 7 ? `${end}-31` : end.slice(0, 10);
  return first <= last ? `${start}/${end}` : null; // the interval must not run backwards
}

function isLatLong(lat, long) {
  return typeof lat === "number" && typeof long === "number" && Math.abs(lat) <= 90 && Math.abs(long) <= 180;
}

// A named place, a GeoCoordinates point, or a GeoShape box ("lat1 long1 lat2 long2", latitude first).
function buildSpatialCoverage(place) {
  if (typeof place === "string") return place;
  if (!place) return null;
  if (place.point) {
    const { lat, long } = place.point;
    if (!isLatLong(lat, long)) return null;
    return { "@type": "Place", name: place.name, geo: { "@type": "GeoCoordinates", latitude: lat, longitude: long } };
  }
  if (Array.isArray(place.box) && place.box.length === 4) {
    const [lat1, long1, lat2, long2] = place.box;
    if (!isLatLong(lat1, long1) || !isLatLong(lat2, long2)) return null;
    return { "@type": "Place", name: place.name, geo: { "@type": "GeoShape", box: place.box.join(" ") } };
  }
  return place.name || null;
}

// Person: name, givenName, familyName, url, ORCID sameAs. Organization: name, url, ROR sameAs.
function buildAgent(agent) {
  if (!agent || !agent.name) return null;
  if (agent.type === "person") {
    return {
      "@type": "Person",
      name: agent.name,
      givenName: agent.givenName,
      familyName: agent.familyName,
      url: isAbsoluteUrl(agent.url) ? agent.url : null,
      sameAs: ORCID.test(agent.orcid || "") ? agent.orcid : null
    };
  }
  return {
    "@type": "Organization",
    name: agent.name,
    url: isAbsoluteUrl(agent.url) ? agent.url : null,
    sameAs: ROR.test(agent.ror || "") ? agent.ror : null
  };
}

// Bare DOIs become resolvable https://doi.org/ URLs; other identifiers stay as given.
function toIdentifier(value) {
  if (typeof value !== "string" || !value.trim()) return null;
  const text = value.trim();
  return DOI.test(text) ? `https://doi.org/${text}` : text;
}

function buildVariable(variable) {
  if (typeof variable === "string") return variable;
  if (!variable || !variable.name) return null;
  return { "@type": "PropertyValue", name: variable.name, unitText: variable.unit, description: variable.description };
}

function buildDistribution(file) {
  // contentUrl is required on every DataDownload.
  if (!file || !isAbsoluteUrl(file.contentUrl)) return null;
  return { "@type": "DataDownload", contentUrl: file.contentUrl, encodingFormat: file.encodingFormat };
}

// A related dataset as a URL, or as a Dataset with its own name and valid description.
function buildRelatedDataset(part) {
  if (isAbsoluteUrl(part)) return part;
  const length = textLength(part && part.description);
  if (!part || !part.name || length < 50 || length > 5000) return null;
  return { "@type": "Dataset", name: part.name, description: part.description.trim(), url: isAbsoluteUrl(part.url) ? part.url : null };
}

function buildDatasetSchema(source) {
  // (a) Page gates: the indexable, self-canonical landing page for this dataset, with the metadata visible.
  if (!passesPageGates(source) || !isAbsoluteUrl(source.url)) return null;

  // (b) Required: name, and a description between 50 and 5000 characters.
  const length = textLength(source.description);
  if (!source.name || length < 50 || length > 5000) return null;

  const identifier = (source.identifiers || []).map(toIdentifier).filter(Boolean);
  const isOwnCitation = (c) => identifier.includes(toIdentifier(c));

  // (c) Recommended properties, only when the source has them; (d) value rules inline.
  const schema = {
    "@context": "https://schema.org",
    "@type": "Dataset",
    "@id": `${source.url}#dataset`,
    name: source.name,
    description: source.description.trim(),
    alternateName: source.alternateNames,
    url: source.url,
    sameAs: (source.sameAs || []).filter(isAbsoluteUrl), // the canonical page when republished
    identifier, // repeat for multiple DOIs or Compact Identifiers
    version: source.version,
    keywords: source.keywords,
    license: isAbsoluteUrl(source.license) ? source.license : null, // a URL for one specific license version
    isAccessibleForFree: typeof source.free === "boolean" ? source.free : null,
    creator: (source.creators || []).map(buildAgent),
    funder: (source.funders || []).map(buildAgent),
    citation: (source.citations || []).filter((c) => !isOwnCitation(c)), // related articles, not the dataset itself
    temporalCoverage: toTemporalCoverage(source.temporalCoverage),
    spatialCoverage: buildSpatialCoverage(source.spatialCoverage),
    variableMeasured: (source.variables || []).map(buildVariable),
    measurementTechnique: source.measurementTechniques,
    isPartOf: (source.partOf || []).map(buildRelatedDataset),
    hasPart: (source.parts || []).map(buildRelatedDataset),
    includedInDataCatalog: source.catalog && source.catalog.name
      ? { "@type": "DataCatalog", name: source.catalog.name, url: isAbsoluteUrl(source.catalog.url) ? source.catalog.url : null }
      : null,
    distribution: (source.files || []).map(buildDistribution)
  };

  // (e) Strip empty values.
  return compact(schema);
}

// Usage:
// const source = {
//   "url": "https://www.example.com/data/charles-river-water-temperature",
//   "canonicalUrl": "https://www.example.com/data/charles-river-water-temperature",
//   "indexable": true,
//   "contentVisible": true,
//   "name": "Charles River Hourly Water Temperature, 2016 to 2025",
//   "description": "Hourly water temperature readings from 12 monitoring stations along the Charles River, Massachusetts, collected from January 2016 to December 2025 and quality controlled in 2026.",
//   "alternateNames": [
//     "CRWT 2016-2025"
//   ],
//   "identifiers": [
//     "10.5555/example.crwt.2026"
//   ],
//   "version": "2026.1",
//   "keywords": [
//     "water temperature",
//     "rivers",
//     "hydrology",
//     "Massachusetts"
//   ],
//   "license": "https://creativecommons.org/licenses/by/4.0/",
//   "free": true,
//   "creators": [
//     {
//       "type": "person",
//       "name": "Josiah Carberry",
//       "givenName": "Josiah",
//       "familyName": "Carberry",
//       "orcid": "https://orcid.org/0000-0002-1825-0097",
//       "url": "https://www.example.com/people/josiah-carberry"
//     }
//   ],
//   "funders": [
//     {
//       "type": "organization",
//       "name": "Example Watershed Foundation",
//       "url": "https://foundation.example.com/",
//       "ror": "https://ror.org/0abcdef12"
//     }
//   ],
//   "citations": [
//     "https://doi.org/10.5555/example.article.2025"
//   ],
//   "temporalCoverage": "2016-01-01/2025-12-31",
//   "spatialCoverage": {
//     "name": "Charles River watershed, Massachusetts",
//     "box": [
//       42.1,
//       -71.6,
//       42.4,
//       -71.0
//     ]
//   },
//   "variables": [
//     {
//       "name": "Water temperature",
//       "unit": "degrees Celsius"
//     }
//   ],
//   "measurementTechniques": [
//     "Submerged thermistor loggers, 60 minute sampling interval"
//   ],
//   "partOf": [
//     "https://www.example.com/data/new-england-river-observations"
//   ],
//   "parts": [
//     "https://www.example.com/data/charles-river-water-temperature/2025"
//   ],
//   "catalog": {
//     "name": "Example River Institute Data Catalog",
//     "url": "https://www.example.com/data/"
//   },
//   "files": [
//     {
//       "contentUrl": "https://www.example.com/data/files/crwt-2016-2025.csv",
//       "encodingFormat": "text/csv"
//     },
//     {
//       "contentUrl": "https://www.example.com/data/files/crwt-2016-2025.parquet",
//       "encodingFormat": "application/vnd.apache.parquet"
//     }
//   ],
//   "sameAs": [
//     "https://catalog.example.org/records/crwt-2016-2025"
//   ]
// };
// const schema = buildDatasetSchema(source); // null when a gate or required property fails

import re
from datetime import date, datetime, timezone

ISO_DATE_TIME = re.compile(r"^(\d{4})-(\d{2})-(\d{2})(?:T(\d{2}):(\d{2})(?::(\d{2}))?(Z|[+-]\d{2}:\d{2})?)?$")
ABSOLUTE_URL = re.compile(r"^https?://[^\s/?#]+\.[^\s/?#]+(?:[/?#]\S*)?$", re.I)


def is_absolute_url(value):
    return isinstance(value, str) and bool(ABSOLUTE_URL.match(value))


# Returns the ISO 8601 string unchanged (timezone offset kept), or None if unparseable.
def to_iso_date_time(value):
    if not isinstance(value, str):
        return None
    text = value.strip()
    m = ISO_DATE_TIME.match(text)
    if not m:
        return None
    y, mo, d = int(m[1]), int(m[2]), int(m[3])
    try:
        date(y, mo, d)
    except ValueError:
        return None
    if int(m[4] or 0) > 23 or int(m[5] or 0) > 59 or int(m[6] or 0) > 59:
        return None
    return text


def to_iso_date(value):
    iso = to_iso_date_time(value)
    return iso[:10] if iso else None


# Drops None, "", [] and {} recursively.
def compact(value):
    if isinstance(value, list):
        items = [c for c in (compact(v) for v in value) if c is not None]
        return items or None
    if isinstance(value, dict):
        out = {k: c for k, c in ((k, compact(v)) for k, v in value.items()) if c is not None}
        return out or None
    return None if value is None or value == "" else value


def passes_page_gates(source):
    if source.get("indexable") is False:
        return False
    if source.get("canonicalUrl") and source.get("url") and source["canonicalUrl"] != source["url"]:
        return False
    return source.get("contentVisible") is not False


ORCID = re.compile(r"^https://orcid\.org/\d{4}-\d{4}-\d{4}-\d{3}[\dX]$")
ROR = re.compile(r"^https://ror\.org/0[a-hj-km-np-tv-z0-9]{6}\d{2}$")
DOI = re.compile(r"^10\.\d{4,9}/\S+$")
PARTIAL_DATE = re.compile(r"^(\d{4})(?:-(\d{2})(?:-(\d{2}))?)?$")


def is_number(value):
    return isinstance(value, (int, float)) and not isinstance(value, bool)


# Formats a number the way JavaScript prints it (42.0 becomes "42").
def js_number(value):
    return str(int(value)) if float(value).is_integer() else repr(value)


def text_length(value):
    return len(value.strip()) if isinstance(value, str) else 0


# "2016", "2016-05", "2016-05-01" or a full ISO 8601 date time.
def to_temporal_point(value):
    m = PARTIAL_DATE.match(value) if isinstance(value, str) else None
    if m and m[3]:
        return to_iso_date(value)
    if m:
        return value if not m[2] or 1 <= int(m[2]) <= 12 else None
    return to_iso_date_time(value) if isinstance(value, str) and "T" in value else None


# ISO 8601 date or interval: "2008", "1950-01-01/2013-12-18", or "2013-12-19/.." for an open end.
def to_temporal_coverage(value):
    if not isinstance(value, str):
        return None
    parts = value.strip().split("/")
    start = parts[0]
    if len(parts) > 2 or not to_temporal_point(start):
        return None
    if len(parts) == 1:
        return start
    end = parts[1]
    if end == "..":
        return f"{start}/.."
    if not to_temporal_point(end):
        return None
    first = f"{start}-01-01" if len(start) == 4 else f"{start}-01" if len(start) == 7 else start[:10]
    last = f"{end}-12-31" if len(end) == 4 else f"{end}-31" if len(end) == 7 else end[:10]
    return f"{start}/{end}" if first <= last else None  # the interval must not run backwards


def is_lat_long(lat, long):
    return is_number(lat) and is_number(long) and abs(lat) <= 90 and abs(long) <= 180


# A named place, a GeoCoordinates point, or a GeoShape box ("lat1 long1 lat2 long2", latitude first).
def build_spatial_coverage(place):
    if isinstance(place, str):
        return place
    if not place:
        return None
    if place.get("point"):
        lat, long = place["point"].get("lat"), place["point"].get("long")
        if not is_lat_long(lat, long):
            return None
        return {"@type": "Place", "name": place.get("name"), "geo": {"@type": "GeoCoordinates", "latitude": lat, "longitude": long}}
    box = place.get("box")
    if isinstance(box, list) and len(box) == 4:
        if not is_lat_long(box[0], box[1]) or not is_lat_long(box[2], box[3]):
            return None
        return {"@type": "Place", "name": place.get("name"), "geo": {"@type": "GeoShape", "box": " ".join(js_number(n) for n in box)}}
    return place.get("name")


# Person: name, givenName, familyName, url, ORCID sameAs. Organization: name, url, ROR sameAs.
def build_agent(agent):
    if not agent or not agent.get("name"):
        return None
    url = agent.get("url") if is_absolute_url(agent.get("url")) else None
    if agent.get("type") == "person":
        return {
            "@type": "Person",
            "name": agent["name"],
            "givenName": agent.get("givenName"),
            "familyName": agent.get("familyName"),
            "url": url,
            "sameAs": agent.get("orcid") if ORCID.match(agent.get("orcid") or "") else None,
        }
    return {
        "@type": "Organization",
        "name": agent["name"],
        "url": url,
        "sameAs": agent.get("ror") if ROR.match(agent.get("ror") or "") else None,
    }


# Bare DOIs become resolvable https://doi.org/ URLs; other identifiers stay as given.
def to_identifier(value):
    if not isinstance(value, str) or not value.strip():
        return None
    text = value.strip()
    return f"https://doi.org/{text}" if DOI.match(text) else text


def build_variable(variable):
    if isinstance(variable, str):
        return variable
    if not variable or not variable.get("name"):
        return None
    return {"@type": "PropertyValue", "name": variable["name"], "unitText": variable.get("unit"), "description": variable.get("description")}


def build_distribution(file):
    # contentUrl is required on every DataDownload.
    if not file or not is_absolute_url(file.get("contentUrl")):
        return None
    return {"@type": "DataDownload", "contentUrl": file["contentUrl"], "encodingFormat": file.get("encodingFormat")}


# A related dataset as a URL, or as a Dataset with its own name and valid description.
def build_related_dataset(part):
    if is_absolute_url(part):
        return part
    if not isinstance(part, dict):
        return None
    length = text_length(part.get("description"))
    if not part.get("name") or length < 50 or length > 5000:
        return None
    url = part.get("url") if is_absolute_url(part.get("url")) else None
    return {"@type": "Dataset", "name": part["name"], "description": part["description"].strip(), "url": url}


def build_dataset_schema(source):
    # (a) Page gates: the indexable, self-canonical landing page for this dataset, with the metadata visible.
    if not passes_page_gates(source) or not is_absolute_url(source.get("url")):
        return None

    # (b) Required: name, and a description between 50 and 5000 characters.
    length = text_length(source.get("description"))
    if not source.get("name") or length < 50 or length > 5000:
        return None

    identifier = [i for i in (to_identifier(v) for v in source.get("identifiers") or []) if i]
    catalog = source.get("catalog") or {}

    # (c) Recommended properties, only when the source has them; (d) value rules inline.
    schema = {
        "@context": "https://schema.org",
        "@type": "Dataset",
        "@id": f"{source['url']}#dataset",
        "name": source["name"],
        "description": source["description"].strip(),
        "alternateName": source.get("alternateNames"),
        "url": source["url"],
        "sameAs": [u for u in source.get("sameAs") or [] if is_absolute_url(u)],  # the canonical page when republished
        "identifier": identifier,  # repeat for multiple DOIs or Compact Identifiers
        "version": source.get("version"),
        "keywords": source.get("keywords"),
        "license": source.get("license") if is_absolute_url(source.get("license")) else None,  # a URL for one specific license version
        "isAccessibleForFree": source.get("free") if isinstance(source.get("free"), bool) else None,
        "creator": [build_agent(a) for a in source.get("creators") or []],
        "funder": [build_agent(a) for a in source.get("funders") or []],
        "citation": [c for c in source.get("citations") or [] if to_identifier(c) not in identifier],  # related articles, not the dataset itself
        "temporalCoverage": to_temporal_coverage(source.get("temporalCoverage")),
        "spatialCoverage": build_spatial_coverage(source.get("spatialCoverage")),
        "variableMeasured": [build_variable(v) for v in source.get("variables") or []],
        "measurementTechnique": source.get("measurementTechniques"),
        "isPartOf": [build_related_dataset(p) for p in source.get("partOf") or []],
        "hasPart": [build_related_dataset(p) for p in source.get("parts") or []],
        "includedInDataCatalog": {
            "@type": "DataCatalog",
            "name": catalog["name"],
            "url": catalog.get("url") if is_absolute_url(catalog.get("url")) else None,
        } if catalog.get("name") else None,
        "distribution": [build_distribution(f) for f in source.get("files") or []],
    }

    # (e) Strip empty values.
    return compact(schema)

{
  "@context": "https://schema.org",
  "@type": "Dataset",
  "@id": "https://www.example.com/data/charles-river-water-temperature#dataset",
  "name": "Charles River Hourly Water Temperature, 2016 to 2025",
  "description": "Hourly water temperature readings from 12 monitoring stations along the Charles River, Massachusetts, collected from January 2016 to December 2025 and quality controlled in 2026.",
  "alternateName": [
    "CRWT 2016-2025"
  ],
  "url": "https://www.example.com/data/charles-river-water-temperature",
  "sameAs": [
    "https://catalog.example.org/records/crwt-2016-2025"
  ],
  "identifier": [
    "https://doi.org/10.5555/example.crwt.2026"
  ],
  "version": "2026.1",
  "keywords": [
    "water temperature",
    "rivers",
    "hydrology",
    "Massachusetts"
  ],
  "license": "https://creativecommons.org/licenses/by/4.0/",
  "isAccessibleForFree": true,
  "creator": [
    {
      "@type": "Person",
      "name": "Josiah Carberry",
      "givenName": "Josiah",
      "familyName": "Carberry",
      "url": "https://www.example.com/people/josiah-carberry",
      "sameAs": "https://orcid.org/0000-0002-1825-0097"
    }
  ],
  "funder": [
    {
      "@type": "Organization",
      "name": "Example Watershed Foundation",
      "url": "https://foundation.example.com/",
      "sameAs": "https://ror.org/0abcdef12"
    }
  ],
  "citation": [
    "https://doi.org/10.5555/example.article.2025"
  ],
  "temporalCoverage": "2016-01-01/2025-12-31",
  "spatialCoverage": {
    "@type": "Place",
    "name": "Charles River watershed, Massachusetts",
    "geo": {
      "@type": "GeoShape",
      "box": "42.1 -71.6 42.4 -71"
    }
  },
  "variableMeasured": [
    {
      "@type": "PropertyValue",
      "name": "Water temperature",
      "unitText": "degrees Celsius"
    }
  ],
  "measurementTechnique": [
    "Submerged thermistor loggers, 60 minute sampling interval"
  ],
  "isPartOf": [
    "https://www.example.com/data/new-england-river-observations"
  ],
  "hasPart": [
    "https://www.example.com/data/charles-river-water-temperature/2025"
  ],
  "includedInDataCatalog": {
    "@type": "DataCatalog",
    "name": "Example River Institute Data Catalog",
    "url": "https://www.example.com/data/"
  },
  "distribution": [
    {
      "@type": "DataDownload",
      "contentUrl": "https://www.example.com/data/files/crwt-2016-2025.csv",
      "encodingFormat": "text/csv"
    },
    {
      "@type": "DataDownload",
      "contentUrl": "https://www.example.com/data/files/crwt-2016-2025.parquet",
      "encodingFormat": "application/vnd.apache.parquet"
    }
  ]
}