"""Generate a ``.spec`` file from a Socrata story JSON."""
from __future__ import annotations
import asyncio
import inspect
import json
import re
import urllib.error
import urllib.parse
import urllib.request
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
import httpx
_DISCOVERY_MAX_WORKERS = 20
from openbb_cli.dispatchers.spec import SPEC_VERSION, _generator_identifier
_CACHED_TOP_CAP = 20
_MAX_RESOLVED_CHOICES = 1000
_NUMERIC_CHOICE_CAP = 50
_NUMERIC_TYPES: frozenset[str] = frozenset({"number", "money", "percent", "double"})
def _coerce_choices_to_type(values: list[Any], data_type: str) -> list[Any]:
"""Cast distinct-value strings to match the column's declared type."""
if data_type in _NUMERIC_TYPES:
coerced: list[Any] = []
for v in values:
try:
num = float(v)
except (TypeError, ValueError):
continue
if num.is_integer():
coerced.append(int(num))
else:
coerced.append(num)
return coerced
if data_type == "checkbox":
coerced_bool: list[Any] = []
for v in values:
if v is None:
continue
if isinstance(v, bool):
coerced_bool.append(v)
elif isinstance(v, str) and v.lower() in {"true", "false"}:
coerced_bool.append(v.lower() == "true")
return coerced_bool
return values
def _cached_top_values(column: dict[str, Any]) -> list[Any]:
"""Pull the items out of a column's ``cachedContents.top`` list."""
cached = column.get("cachedContents") or {}
top = cached.get("top") or []
values: list[Any] = []
for entry in top:
if isinstance(entry, dict):
item = entry.get("item")
if item is not None:
values.append(item)
elif entry is not None:
values.append(entry)
return values
def _is_categorical_column(column: dict[str, Any]) -> bool:
"""Return True when querying for distinct values is worthwhile."""
return (column.get("dataTypeName") or "text") in ("text", "html")
async def afetch_column_distinct_values(
client: httpx.AsyncClient,
host: str,
uid: str,
field_name: str,
*,
limit: int = _MAX_RESOLVED_CHOICES,
) -> list[Any] | None:
"""Resolve a column's full distinct-value set via SoQL (async)."""
query = urllib.parse.urlencode(
{
"$select": field_name,
"$group": field_name,
"$order": field_name,
"$limit": str(limit + 1),
}
)
url = f"{host.rstrip('/')}/resource/{uid}.json?{query}"
try:
response = await client.get(url)
response.raise_for_status()
payload = response.json()
except (httpx.HTTPError, ValueError, OSError):
return None
if not isinstance(payload, list) or len(payload) > limit:
return None
out: list[Any] = []
for row in payload:
if not isinstance(row, dict):
continue
value = row.get(field_name)
if value is not None:
out.append(value)
return out
def _column_choices(
column: dict[str, Any],
*,
host: str | None = None,
uid: str | None = None,
resolver: Any = None,
choices_cache: dict[tuple[str, str], list[Any] | None] | None = None,
) -> list[Any]:
"""Return the closed ``choices`` set for a column, or ``[]`` if unbounded."""
field_name = column.get("fieldName") or column.get("name")
if (
choices_cache is not None
and host
and uid
and field_name
and (uid, field_name) in choices_cache
):
cached = choices_cache[(uid, field_name)]
if cached is None:
return []
type_name = column.get("dataTypeName") or ""
return _coerce_choices_to_type(cached, type_name)
if not host or not uid:
return _cached_top_values(column)
if not field_name:
return []
if resolver is None:
return _cached_top_values(column)
fetch = resolver
type_name = column.get("dataTypeName") or ""
cap = _NUMERIC_CHOICE_CAP if type_name in _NUMERIC_TYPES else _MAX_RESOLVED_CHOICES
resolved = fetch(host, uid, field_name, limit=cap)
if resolved is not None:
return _coerce_choices_to_type(resolved, type_name)
cached = column.get("cachedContents") or {}
if 0 < len(cached.get("top") or []) < _CACHED_TOP_CAP:
return _cached_top_values(column)
return []
def _column_help(column: dict[str, Any]) -> str:
"""Compose the per-column help text shown under ``--
=``."""
parts: list[str] = []
description = column.get("description")
if description:
parts.append(description.strip().rstrip(".") + ".")
cached = column.get("cachedContents") or {}
top = cached.get("top") or []
if top and len(top) >= _CACHED_TOP_CAP:
sample = []
for entry in top[:5]:
item = entry.get("item") if isinstance(entry, dict) else entry
if item is not None:
sample.append(str(item))
if sample:
parts.append(f"Examples: {', '.join(sample)}.")
smallest = cached.get("smallest")
largest = cached.get("largest")
if smallest is not None and largest is not None and len(top) >= _CACHED_TOP_CAP:
parts.append(f"Range: {smallest} – {largest}.")
return " ".join(parts)
def _column_param_type(column: dict[str, Any]) -> str:
"""Map a Socrata ``dataTypeName`` to a spec ``type`` value."""
type_name = column.get("dataTypeName") or "text"
if type_name in ("number", "money", "double", "percent"):
return "number"
if type_name == "checkbox":
return "boolean"
return "string"
_DATE_COLUMN_TYPES: frozenset[str] = frozenset({"calendar_date", "date"})
_TIME_AXIS_NAME_PATTERNS: tuple[str, ...] = (
"month_year",
"year_month",
"yearmonth",
"monthyear",
"as_of_date",
"as_of",
"asof",
"observation_date",
"report_date",
"release_date",
"effective_date",
"date",
"datetime",
"timestamp",
"time",
"year",
"quarter",
"month",
"week",
"period",
)
def _is_date_column(column: dict[str, Any]) -> bool:
"""Return True when the column's data type calls for range-style filtering."""
return (column.get("dataTypeName") or "") in _DATE_COLUMN_TYPES
def _is_time_axis_named(field_name: str) -> bool:
"""Return True when the column's name suggests it carries a time value."""
name = (field_name or "").lower()
return name in _TIME_AXIS_NAME_PATTERNS
def _detect_time_axis_column(meta: dict[str, Any]) -> str | None:
"""Pick the column to sort by descending for most-recent semantics."""
for col in meta.get("columns") or []:
if not isinstance(col, dict):
continue
if _is_date_column(col):
field_name = col.get("fieldName") or col.get("name")
if field_name:
return field_name
candidates: list[str] = []
for col in meta.get("columns") or []:
if not isinstance(col, dict):
continue
field_name = col.get("fieldName") or col.get("name")
if field_name and _is_time_axis_named(field_name):
candidates.append(field_name)
if not candidates:
return None
candidates.sort(key=len, reverse=True)
return candidates[0]
def _to_plain_date(value: Any) -> Any:
"""Trim the time component off a Socrata datetime string."""
if not isinstance(value, str):
return value
if "T" in value and len(value) >= 10 and value[4] == "-" and value[7] == "-":
return value.split("T", 1)[0]
return value
def _date_range_parameters(column: dict[str, Any]) -> list[dict[str, Any]]:
"""Emit a uniform ``start_date`` / ``end_date`` pair for a date column."""
field_name = column.get("fieldName") or column.get("name")
cached = column.get("cachedContents") or {}
smallest = _to_plain_date(cached.get("smallest"))
largest = _to_plain_date(cached.get("largest"))
description = column.get("description") or ""
base_help = description.strip().rstrip(".")
range_text = (
f" Dataset spans {smallest} to {largest}." if smallest and largest else ""
)
return [
{
"name": "start_date",
"in": "query",
"type": "string",
"is_list": False,
"required": False,
"default": None,
"choices": [],
"example": smallest,
"help": (
f"Earliest date (YYYY-MM-DD) to include — filters on ``{field_name}``."
f" Omit to fetch from the dataset's earliest record."
f"{(' ' + base_help + '.') if base_help else ''}"
f"{range_text}"
).strip(),
"providers": [],
"_socrata_op": "date_min",
"_socrata_column": field_name,
},
{
"name": "end_date",
"in": "query",
"type": "string",
"is_list": False,
"required": False,
"default": None,
"choices": [],
"example": largest,
"help": (
f"Latest date (YYYY-MM-DD) to include — filters on ``{field_name}``."
f" Omit to fetch through the dataset's latest record."
f"{(' ' + base_help + '.') if base_help else ''}"
f"{range_text}"
).strip(),
"providers": [],
"_socrata_op": "date_max",
"_socrata_column": field_name,
},
]
def _build_query_parameters(
meta: dict[str, Any],
*,
host: str | None = None,
uid: str | None = None,
resolver: Any = None,
choices_cache: dict[tuple[str, str], list[Any] | None] | None = None,
) -> list[dict[str, Any]]:
"""Build the per-dataset query parameter list."""
parameters: list[dict[str, Any]] = []
date_range_emitted = False
for column in meta.get("columns") or []:
if not isinstance(column, dict):
continue
field_name = column.get("fieldName") or column.get("name")
if not field_name:
continue
if _is_date_column(column):
if not date_range_emitted:
parameters.extend(_date_range_parameters(column))
date_range_emitted = True
continue
parameters.append(
{
"name": field_name,
"in": "query",
"type": _column_param_type(column),
"is_list": False,
"required": False,
"default": None,
"choices": _column_choices(
column,
host=host,
uid=uid,
resolver=resolver,
choices_cache=choices_cache,
),
"example": None,
"help": _column_help(column),
"providers": [],
}
)
parameters.extend(
[
{
"name": "limit",
"in": "query",
"type": "integer",
"is_list": False,
"required": False,
"default": 1000,
"choices": [],
"example": None,
"help": (
"Max rows returned. Socrata's API default is 1000; up to "
"50000 per request without pagination."
),
"providers": [],
"wire_name": "$limit",
},
{
"name": "offset",
"in": "query",
"type": "integer",
"is_list": False,
"required": False,
"default": None,
"choices": [],
"example": None,
"help": "Row offset for pagination — pair with limit.",
"providers": [],
"wire_name": "$offset",
},
]
)
return parameters
_SOCRATA_TYPE_MAP: dict[str, dict[str, Any]] = {
"text": {"type": "string"},
"html": {"type": "string"},
"number": {"type": "number"},
"money": {"type": "number"},
"percent": {"type": "number"},
"double": {"type": "number"},
"checkbox": {"type": "boolean"},
"calendar_date": {"type": "string", "format": "date"},
"date": {"type": "string", "format": "date"},
"url": {
"type": "object",
"properties": {
"url": {"type": "string"},
"description": {"type": "string"},
},
},
"email": {"type": "string"},
"phone": {"type": "object"},
"point": {"type": "object"},
"location": {"type": "object"},
"multipoint": {"type": "object"},
"line": {"type": "object"},
"multiline": {"type": "object"},
"polygon": {"type": "object"},
"multipolygon": {"type": "object"},
"photo": {"type": "object"},
"document": {"type": "object"},
}
def _slugify(name: str) -> str:
"""Translate a free-form dataset name to a snake_case identifier."""
cleaned = re.sub(r"[^0-9a-zA-Z]+", "_", name).strip("_").lower()
if not cleaned:
cleaned = "dataset"
if cleaned[0].isdigit():
cleaned = f"_{cleaned}"
return cleaned
_PREFIX_STOP_TOKENS: frozenset[str] = frozenset(
{"by", "for", "of", "in", "at", "to", "with", "and", "the"}
)
def _common_token_prefix(token_lists: list[list[str]]) -> list[str]:
"""Return the longest leading token sequence shared by every input."""
if not token_lists:
return []
shortest = min(len(tl) for tl in token_lists)
common: list[str] = []
for i in range(shortest):
token = token_lists[0][i]
if all(tl[i] == token for tl in token_lists):
common.append(token)
else:
break
return common
def _longest_common_token_run(token_lists: list[list[str]]) -> list[str]:
"""Return the longest contiguous token run shared by every input list."""
if not token_lists:
return []
first = token_lists[0]
n = len(first)
others = token_lists[1:]
for length in range(n, 0, -1):
for start in range(n - length + 1):
candidate = first[start : start + length]
if all(
any(
tl[i : i + length] == candidate for i in range(len(tl) - length + 1)
)
for tl in others
):
return candidate
return []
def _strip_stop_tokens(tokens: list[str]) -> list[str]:
"""Trim leading and trailing connector words from a token list."""
out = list(tokens)
while out and out[-1] in _PREFIX_STOP_TOKENS:
out.pop()
while out and out[0] in _PREFIX_STOP_TOKENS:
out.pop(0)
return out
_MIN_SHARED_RUN = 2
def _assign_router_namespaces(
slugs: dict[str, str],
) -> dict[str, tuple[str | None, str]]:
"""Group dataset slugs into ``(router, command)`` pairs.
Parameters
----------
slugs : dict
Mapping of dataset UID to slugified dataset name.
Returns
-------
dict
Mapping of dataset UID to ``(router_or_None, command)`` tuple.
"""
if not slugs:
return {}
token_lists: dict[str, list[str]] = {
uid: slug.split("_") for uid, slug in slugs.items()
}
uids = list(slugs.keys())
parent: dict[str, str] = {uid: uid for uid in uids}
def find(u: str) -> str:
while parent[u] != u:
parent[u] = parent[parent[u]]
u = parent[u]
return u
def union(u: str, v: str) -> None:
ru, rv = find(u), find(v)
if ru != rv:
parent[ru] = rv
for i in range(len(uids)):
for j in range(i + 1, len(uids)):
run = _longest_common_token_run(
[token_lists[uids[i]], token_lists[uids[j]]]
)
if len(run) >= _MIN_SHARED_RUN:
union(uids[i], uids[j])
groups: dict[str, list[str]] = {}
for uid in uids:
groups.setdefault(find(uid), []).append(uid)
out: dict[str, tuple[str | None, str]] = {}
for group_uids in groups.values():
if len(group_uids) == 1:
uid = group_uids[0]
out[uid] = (None, slugs[uid])
continue
group_tokens = [token_lists[u] for u in group_uids]
router_tokens = _strip_stop_tokens(_longest_common_token_run(group_tokens))
if not router_tokens:
for uid in group_uids:
out[uid] = (None, slugs[uid])
continue
router = "_".join(router_tokens)
run_len = len(router_tokens)
member_commands = {
uid: _command_tokens_around_run(token_lists[uid], router_tokens, run_len)
for uid in group_uids
}
if all(not c for c in member_commands.values()):
for uid in group_uids:
out[uid] = (None, slugs[uid])
continue
for uid in group_uids:
tokens = member_commands[uid]
command = "_".join(tokens) if tokens else "query"
out[uid] = (router, command)
return out
def _command_tokens_around_run(
tokens: list[str], router_tokens: list[str], run_len: int
) -> list[str]:
"""Return ``tokens`` minus the first occurrence of ``router_tokens``."""
for i in range(len(tokens) - run_len + 1):
if tokens[i : i + run_len] == router_tokens:
return tokens[:i] + tokens[i + run_len :]
return list(tokens)
def _walk_for_dataset_uids(node: Any, sink: set[str]) -> None:
"""Recursively collect every ``datasetUid`` value in a JSON tree."""
if isinstance(node, dict):
for key, value in node.items():
if key == "datasetUid" and isinstance(value, str):
sink.add(value)
else:
_walk_for_dataset_uids(value, sink)
elif isinstance(node, list):
for item in node:
_walk_for_dataset_uids(item, sink)
def extract_dataset_uids(story: dict[str, Any]) -> list[str]:
"""Return the unique dataset UIDs referenced by a story, in stable order."""
sink: set[str] = set()
_walk_for_dataset_uids(story, sink)
return sorted(sink)
_STORY_URL_RE = re.compile(
r"^(?Phttps?://[^/]+/stories/s/[a-z0-9]{4}-[a-z0-9]{4})(?P.*)$"
)
_SINGLE_VIEW_URL_RE = re.compile(
r"^(?Phttps?://[^/]+)"
r"/(?:resource|api/views)/(?P[a-z0-9]{4}-[a-z0-9]{4})"
r"(?:\.json)?(?:\?.*)?$"
)
def _direct_dataset_seed(source: str) -> tuple[str, str] | None:
"""Detect ``/resource/`` or ``/api/views/`` URLs and return ``(host, uid)``."""
match = _SINGLE_VIEW_URL_RE.match(source)
if match:
return match.group("host"), match.group("uid")
return None
def _normalize_story_url(source: str) -> str:
"""Append ``.json`` when the user pastes the browser-facing story URL."""
match = _STORY_URL_RE.match(source)
if not match:
return source
rest = match.group("rest")
if rest.startswith(".json") or rest.startswith("?") and ".json" in rest:
return source
if not rest:
return f"{match.group('base')}.json"
if rest.startswith("?"):
return f"{match.group('base')}.json{rest}"
return source
async def _aread_json_url_or_path(source: str, *, timeout: float = 15.0) -> Any:
"""Load JSON from an HTTP URL (async via httpx) or local file path (sync)."""
if source.startswith(("http://", "https://")):
async with httpx.AsyncClient(timeout=httpx.Timeout(timeout)) as client:
response = await client.get(source)
response.raise_for_status()
return response.json()
return json.loads(Path(source).read_text())
def _story_host(story_source: str, story: dict[str, Any]) -> str:
"""Pick the portal base URL the story's datasets live on."""
domain = (story.get("dataSource") or {}).get("domainCName")
if isinstance(domain, str) and domain:
return f"https://{domain}".rstrip("/")
if story_source.startswith(("http://", "https://")):
parsed = urllib.parse.urlparse(story_source)
return f"{parsed.scheme}://{parsed.netloc}"
raise ValueError(
"story has no ``dataSource.domainCName`` and the source is a local "
"file — pass an HTTP URL or add the host to the story document."
)
async def _afetch_json(client: httpx.AsyncClient, url: str) -> Any:
"""Async GET + JSON-decode, raising on HTTP errors."""
response = await client.get(url)
response.raise_for_status()
return response.json()
async def _amaybe_call(callable_: Any, *args: Any, **kwargs: Any) -> Any:
"""Invoke ``callable_`` and ``await`` the result if it's a coroutine."""
result = callable_(*args, **kwargs)
if inspect.isawaitable(result):
return await result
return result
async def _aone_metadata(
client: httpx.AsyncClient,
host: str,
uid: str,
fetcher: Any,
) -> dict[str, Any] | None:
"""Fetch one view's metadata, returning ``None`` on any failure."""
if fetcher is None or fetcher is afetch_dataset_metadata:
try:
return await afetch_dataset_metadata(client, host, uid)
except (httpx.HTTPError, ValueError, OSError):
return None
try:
return await _amaybe_call(fetcher, host, uid)
except (httpx.HTTPError, OSError, ValueError):
return None
async def _afetch_metadata_batch(
client: httpx.AsyncClient,
host: str,
uids: list[str],
fetcher: Any,
) -> dict[str, dict[str, Any] | None]:
"""Concurrently fetch metadata for many UIDs."""
if not uids:
return {}
sem = asyncio.Semaphore(_DISCOVERY_MAX_WORKERS)
async def _one(uid: str) -> dict[str, Any] | None:
async with sem:
return await _aone_metadata(client, host, uid, fetcher)
results = await asyncio.gather(*(_one(uid) for uid in uids))
return dict(zip(uids, results))
async def _aone_choice(
client: httpx.AsyncClient,
host: str,
uid: str,
field: str,
cap: int,
resolver: Any,
) -> list[Any] | None:
"""Fetch one column's distinct values, ``None`` on overflow / error."""
if resolver is None or resolver is afetch_column_distinct_values:
return await afetch_column_distinct_values(client, host, uid, field, limit=cap)
try:
return await _amaybe_call(resolver, host, uid, field, limit=cap)
except (httpx.HTTPError, OSError, ValueError):
return None
async def _abuild_discovery(
story_source: str,
host_override: str | None,
fetcher: Any,
choice_resolver: Any,
direct_seed: tuple[str, str] | None = None,
) -> tuple[
dict[str, Any],
str,
dict[str, dict[str, Any]],
dict[str, str],
list[tuple[str, str]],
dict[tuple[str, str], list[Any] | None],
]:
"""Run the full async discovery pipeline end-to-end."""
async with httpx.AsyncClient(
timeout=httpx.Timeout(30.0),
limits=httpx.Limits(
max_connections=_DISCOVERY_MAX_WORKERS * 2,
max_keepalive_connections=_DISCOVERY_MAX_WORKERS,
),
) as client:
if direct_seed is not None:
seed_host, seed_uid = direct_seed
story = {"_synthetic_single_view": True, "uid": seed_uid}
host = host_override or seed_host
uids = [seed_uid]
else:
story = await _aload_story(client, story_source)
host = host_override or _story_host(story_source, story)
uids = extract_dataset_uids(story)
metas: dict[str, dict[str, Any]] = {}
slugs: dict[str, str] = {}
skipped: list[tuple[str, str]] = []
pending: list[str] = list(dict.fromkeys(uids))
processed: set[str] = set()
while pending:
wave = [uid for uid in pending if uid not in processed]
pending = []
if not wave: # pragma: no cover
break
processed.update(wave)
wave_metas = await _afetch_metadata_batch(client, host, wave, fetcher)
for uid, meta in wave_metas.items():
if meta is None:
skipped.append((uid, "unreachable"))
continue
asset_type = meta.get("assetType")
if asset_type in {"dataset", "filter"}:
metas[uid] = meta
slugs[uid] = _slugify(meta.get("name") or uid)
else:
skipped.append((uid, asset_type or "unknown"))
backing = _backing_dataset_uid(meta)
if backing and backing not in processed:
pending.append(backing)
choices_cache = await _aprobe_choices_batch(
client, host, metas, choice_resolver
)
return story, host, metas, slugs, skipped, choices_cache
async def _aload_story(client: httpx.AsyncClient, story_source: str) -> dict[str, Any]:
"""Async version of ``_aread_json_url_or_path`` that reuses the client."""
if story_source.startswith(("http://", "https://")):
response = await client.get(story_source)
response.raise_for_status()
return response.json()
return json.loads(Path(story_source).read_text())
async def _aprobe_choices_batch(
client: httpx.AsyncClient,
host: str,
metas: dict[str, dict[str, Any]],
resolver: Any,
) -> dict[tuple[str, str], list[Any] | None]:
"""Concurrently probe distinct-value choices for every column on every dataset."""
tasks: list[tuple[str, str, int]] = []
for uid, meta in metas.items():
for col in meta.get("columns") or []:
if not isinstance(col, dict):
continue
field = col.get("fieldName") or col.get("name")
if not field:
continue
if _is_date_column(col):
continue
type_name = col.get("dataTypeName") or "text"
if type_name not in _NUMERIC_TYPES and type_name not in {
"text",
"html",
"checkbox",
}:
continue
cap = (
_NUMERIC_CHOICE_CAP
if type_name in _NUMERIC_TYPES
else _MAX_RESOLVED_CHOICES
)
tasks.append((uid, field, cap))
if not tasks:
return {}
sem = asyncio.Semaphore(_DISCOVERY_MAX_WORKERS)
async def _one(
uid: str, field: str, cap: int
) -> tuple[tuple[str, str], list[Any] | None]:
async with sem:
value = await _aone_choice(client, host, uid, field, cap, resolver)
return (uid, field), value
results = await asyncio.gather(
*(_one(uid, field, cap) for uid, field, cap in tasks)
)
return dict(results)
async def afetch_dataset_metadata(
client: httpx.AsyncClient, host: str, uid: str
) -> dict[str, Any]:
"""Fetch a Socrata view's metadata JSON (async)."""
url = f"{host.rstrip('/')}/api/views/{uid}.json"
response = await client.get(url)
response.raise_for_status()
return response.json()
def _backing_dataset_uid(meta: dict[str, Any]) -> str | None:
"""Return the UID of a non-dataset view's backing dataset, or ``None``."""
backing = meta.get("modifyingViewUid")
if isinstance(backing, str) and backing:
return backing
query = meta.get("query")
if isinstance(query, dict):
source = query.get("viewSourceId")
if isinstance(source, str) and source:
return source
return None
def _column_to_schema(column: dict[str, Any]) -> dict[str, Any]:
"""Translate one Socrata column descriptor into a JSON-schema fragment."""
type_name = column.get("dataTypeName") or "text"
schema = dict(_SOCRATA_TYPE_MAP.get(type_name, {"type": "string"}))
description = column.get("description")
if description:
cleaned = description.strip()
if cleaned:
schema["description"] = cleaned
fmt = column.get("format")
if isinstance(fmt, dict) and fmt:
rendering_hints = {
k: v
for k, v in fmt.items()
if k
in {
"precisionStyle",
"currency",
"decimalSeparator",
"groupSeparator",
"noCommas",
"precision",
"view",
"mask",
}
}
if rendering_hints:
schema["socrata_format"] = rendering_hints
return schema
def _build_response_schema(meta: dict[str, Any]) -> dict[str, Any]:
"""Build a row-shape schema from a dataset's column metadata."""
columns = meta.get("columns") or []
properties: dict[str, dict[str, Any]] = {}
for column in columns:
if not isinstance(column, dict):
continue
field_name = column.get("fieldName") or column.get("name")
if not field_name:
continue
properties[field_name] = _column_to_schema(column)
return {
"type": "object",
"properties": {
"results": {
"type": "array",
"items": {"type": "object", "properties": properties},
}
},
}
def _build_command(
host: str,
uid: str,
meta: dict[str, Any],
*,
choice_resolver: Any = None,
choices_cache: dict[tuple[str, str], list[Any] | None] | None = None,
) -> dict[str, Any]:
"""Synthesize one spec command entry for a dataset."""
name = (meta.get("name") or uid).strip()
description = (meta.get("description") or "").strip()
summary = description or name
parameters = _build_query_parameters(
meta,
host=host,
uid=uid,
resolver=choice_resolver,
choices_cache=choices_cache,
)
time_axis = _detect_time_axis_column(meta)
return {
"url_path": f"/resource/{uid}.json",
"method": "get",
"description": summary,
"parameters": parameters,
"providers": [],
"request_body_schema": None,
"request_body_schemas": {},
"response_schema": _build_response_schema(meta),
"response_schemas": {},
"_socrata_time_axis": time_axis,
"_socrata_primary_item": _primary_item_column(parameters, time_axis),
}
def _primary_item_column(
parameters: list[dict[str, Any]], time_axis: str | None
) -> str | None:
"""Pick the column to use as the per-item key for ``limit`` semantics."""
excluded: set[str] = {"limit", "offset"}
if time_axis:
excluded.add(time_axis)
candidates: list[tuple[int, str]] = []
for p in parameters:
if not isinstance(p, dict):
continue
if p.get("_socrata_op") in {"date_min", "date_max"}:
continue
name = p.get("name")
if not name or name in excluded:
continue
if _is_time_axis_named(name):
continue
if p.get("type") != "string":
continue
choices = p.get("choices") or []
if len(choices) < 3:
continue
candidates.append((len(choices), name))
if not candidates:
return None
candidates.sort(key=lambda pair: -pair[0])
return candidates[0][1]
def build_socrata_spec(
story_source: str,
*,
host_override: str | None = None,
fetch: Any = None,
choice_resolver: Any = None,
) -> dict[str, Any]:
"""Build a complete spec document from a Socrata story URL or local path.
Parameters
----------
story_source : str
A story URL or a path to a previously-downloaded story JSON file.
host_override : str, optional
Portal base URL to use instead of the one inferred from the story.
fetch : callable, optional
Override the metadata fetcher — signature ``(host, uid) -> dict``.
choice_resolver : callable, optional
Override the per-column distinct-value resolver.
Returns
-------
dict
A spec document conforming to ``SpecDocument``.
"""
story_source = _normalize_story_url(story_source)
direct_seed = _direct_dataset_seed(story_source)
story, host, metas, slugs, skipped, choices_cache = asyncio.run(
_abuild_discovery(
story_source,
host_override,
fetch,
choice_resolver,
direct_seed=direct_seed,
)
)
assignments = _assign_router_namespaces(slugs)
commands: dict[str, dict[str, Any]] = {}
seen_paths: set[str] = set()
for uid, (router, command) in assignments.items():
path = command if router is None else f"{router}.{command}"
if path in seen_paths:
path = f"{path}_{uid.replace('-', '')}"
seen_paths.add(path)
commands[path] = _build_command(
host,
uid,
metas[uid],
choice_resolver=choice_resolver,
choices_cache=choices_cache,
)
return {
"version": SPEC_VERSION,
"generated_at": datetime.now(UTC).isoformat(),
"generator": _generator_identifier(),
"source_url": story_source,
"api_version": "socrata-soda-2.1",
"base_url": host,
"api_prefix": "",
"commands": commands,
"routers": _build_routers(commands),
"reference": _build_reference(commands),
"_socrata": {
"story_uid": story.get("uid"),
"dataset_count": len(commands),
"skipped": [{"uid": u, "asset_type": t} for u, t in skipped],
},
}
def _build_routers(commands: dict[str, dict[str, Any]]) -> dict[str, str]:
"""Mark every namespace prefix as ``menu`` and every leaf as ``command``."""
out: dict[str, str] = {}
for cmd_path in commands:
parts = cmd_path.split(".")
for i in range(1, len(parts)):
out[".".join(parts[:i])] = "menu"
out[cmd_path] = "command"
return out
def _build_reference(
commands: dict[str, dict[str, Any]],
) -> dict[str, dict[str, dict[str, str]]]:
"""Build the slash-style reference index the REPL renders menu help from."""
paths: dict[str, dict[str, str]] = {}
routers: dict[str, dict[str, str]] = {}
for cmd_path, cmd_spec in commands.items():
parts = cmd_path.split(".")
slash_path = "/" + "/".join(parts)
paths[slash_path] = {"description": cmd_spec.get("description", "")}
for i in range(1, len(parts)):
slash_router = "/" + "/".join(parts[:i]) + "/"
routers.setdefault(slash_router, {"description": ""})
return {"paths": paths, "routers": routers}