diff --git a/README.md b/README.md index 17a4bc2..878280b 100644 --- a/README.md +++ b/README.md @@ -10,6 +10,7 @@ CivicSwarm stores proposals, comments, resident profiles, preference signals, an - summarizes discussion by detected language - routes proposal fragments to relevant residents by geography, interests, experience, and language - aggregates preferences with optional differential-privacy noise +- ingests offline capture batches from mobile, SMS, and field canvassing with audit-friendly idempotency - exports a civic brief with provenance and audit trail ## Stack @@ -50,10 +51,12 @@ Environment variables: - `POST /residents` - `POST /proposals/{proposal_id}/comments` - `POST /proposals/{proposal_id}/preferences` +- `POST /capture-batches` - `GET /proposals/{proposal_id}/route` - `GET /proposals/{proposal_id}/dashboard` - `GET /proposals/{proposal_id}/brief` - `GET /proposals/{proposal_id}/ledger` +- `GET /capture-batches/{batch_key}` ## Testing diff --git a/civicswarm/api.py b/civicswarm/api.py index 560a568..7398eda 100644 --- a/civicswarm/api.py +++ b/civicswarm/api.py @@ -1,6 +1,8 @@ from __future__ import annotations +from datetime import datetime from pathlib import Path +from typing import Annotated, Literal from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field @@ -38,6 +40,61 @@ class PreferenceCreate(BaseModel): channel: str = "mobile" +class BatchResident(BaseModel): + kind: Literal["resident"] = "resident" + resident_key: str + geography: str = "" + interests: list[str] = Field(default_factory=list) + lived_experience: list[str] = Field(default_factory=list) + languages: list[str] = Field(default_factory=list) + + +class BatchComment(BaseModel): + kind: Literal["comment"] = "comment" + proposal_id: int + text: str + resident_key: str | None = None + channel: str = "mobile" + geography: str = "" + metadata: dict = Field(default_factory=dict) + + +class BatchPreference(BaseModel): + kind: Literal["preference"] = "preference" + proposal_id: int + resident_key: str + score: float + channel: str = "mobile" + + +class BatchLedger(BaseModel): + kind: Literal["ledger"] = "ledger" + proposal_id: int + entry_kind: str + payload: dict = Field(default_factory=dict) + + +class BatchProposalVersion(BaseModel): + kind: Literal["proposal_version"] = "proposal_version" + proposal_id: int + rationale: str = "" + objections: list[str] = Field(default_factory=list) + resolution_status: str = "open" + + +CaptureItem = Annotated[ + BatchResident | BatchComment | BatchPreference | BatchLedger | BatchProposalVersion, + Field(discriminator="kind"), +] + + +class CaptureBatchCreate(BaseModel): + batch_key: str + source: str = "mobile" + captured_at: datetime | None = None + items: list[CaptureItem] = Field(default_factory=list) + + def create_app(db_path: str | Path = ":memory:") -> FastAPI: service = CivicSwarmService.from_path(db_path) app = FastAPI(title="CivicSwarm", version="0.1.0") @@ -62,6 +119,17 @@ def create_app(db_path: str | Path = ":memory:") -> FastAPI: def create_preference(proposal_id: int, payload: PreferenceCreate): return service.submit_preference(proposal_id, payload.resident_key, payload.score, payload.channel) + @app.post("/capture-batches") + def ingest_capture_batch(payload: CaptureBatchCreate): + return service.ingest_capture_batch(payload.batch_key, payload.source, [item.model_dump() for item in payload.items], payload.captured_at) + + @app.get("/capture-batches/{batch_key}") + def get_capture_batch(batch_key: str): + try: + return service.get_capture_batch(batch_key) + except KeyError as exc: + raise HTTPException(status_code=404, detail=str(exc)) from exc + @app.get("/proposals/{proposal_id}/route") def route_proposal(proposal_id: int, top_n: int = 3): return service.route_proposal(proposal_id, top_n=top_n) diff --git a/civicswarm/db.py b/civicswarm/db.py index fce5ba1..8cb2b87 100644 --- a/civicswarm/db.py +++ b/civicswarm/db.py @@ -18,6 +18,7 @@ from sqlalchemy import ( create_engine, inspect, ) +from sqlalchemy.engine import Engine def utcnow() -> datetime: @@ -103,6 +104,21 @@ routes = Table( Column("created_at", DateTime(timezone=True), nullable=False, default=utcnow), ) +capture_batches = Table( + "capture_batches", + metadata, + Column("id", Integer, primary_key=True), + Column("batch_key", String, nullable=False, unique=True), + Column("source", String, nullable=False), + Column("captured_at", DateTime(timezone=True), nullable=False, default=utcnow), + Column("item_count", Integer, nullable=False, default=0), + Column("status", String, nullable=False, default="received"), + Column("payload_json", Text, nullable=False, default="{}"), + Column("results_json", Text, nullable=False, default="[]"), + Column("created_at", DateTime(timezone=True), nullable=False, default=utcnow), + Column("ingested_at", DateTime(timezone=True), nullable=True), +) + ledger_entries = Table( "ledger_entries", metadata, @@ -116,7 +132,7 @@ ledger_entries = Table( @dataclass(frozen=True) class Database: - engine: object + engine: Engine def create_database(db_url: str): diff --git a/civicswarm/service.py b/civicswarm/service.py index 648e851..9b7f8f7 100644 --- a/civicswarm/service.py +++ b/civicswarm/service.py @@ -23,7 +23,9 @@ from .analysis import ( tokenize, ) from .db import ( + capture_batches, comments, + Database, create_database, ledger_entries, preferences, @@ -60,6 +62,7 @@ def _normalize_terms(values: Iterable[str]) -> set[str]: class CivicSwarmService: db_url: str secret: str = "civicswarm" + db: Database | None = None def __post_init__(self) -> None: self.db = create_database(self.db_url) @@ -167,29 +170,8 @@ class CivicSwarmService: metadata: dict[str, Any] | None = None, ) -> dict[str, Any]: metadata = metadata or {} - language = detect_language(text) with self.db.engine.begin() as conn: - result = conn.execute( - insert(comments).values( - proposal_id=proposal_id, - resident_key=self._token(resident_key), - channel=channel, - geography=geography, - language=language, - text=text, - metadata_json=_json(metadata), - created_at=_now(), - ) - ) - comment_id = result.inserted_primary_key[0] - conn.execute( - insert(ledger_entries).values( - proposal_id=proposal_id, - kind="comment_received", - payload_json=_json({"comment_id": comment_id, "channel": channel, "language": language}), - created_at=_now(), - ) - ) + comment_id = self._insert_comment_row(conn, proposal_id, text, channel, resident_key, geography, metadata) return self.get_comment(comment_id) def get_comment(self, comment_id: int) -> dict[str, Any]: @@ -201,26 +183,101 @@ class CivicSwarmService: def submit_preference(self, proposal_id: int, resident_key: str, score: float, channel: str = "mobile") -> dict[str, Any]: with self.db.engine.begin() as conn: - result = conn.execute( - insert(preferences).values( - proposal_id=proposal_id, - resident_key=self._token(resident_key) or resident_key, - score=float(score), - channel=channel, - created_at=_now(), - ) - ) - preference_id = result.inserted_primary_key[0] - conn.execute( - insert(ledger_entries).values( - proposal_id=proposal_id, - kind="preference_submitted", - payload_json=_json({"preference_id": preference_id, "channel": channel}), - created_at=_now(), - ) - ) + preference_id = self._insert_preference_row(conn, proposal_id, resident_key, score, channel) return {"id": preference_id, "proposal_id": proposal_id, "resident_key": resident_key, "score": float(score), "channel": channel} + def ingest_capture_batch(self, batch_key: str, source: str, items: list[dict[str, Any]], captured_at: datetime | None = None) -> dict[str, Any]: + captured_at = captured_at or _now() + payload = {"batch_key": batch_key, "source": source, "captured_at": captured_at.isoformat(), "items": items} + + with self.db.engine.begin() as conn: + existing = conn.execute(select(capture_batches).where(capture_batches.c.batch_key == batch_key)).mappings().first() + if existing is not None: + return self._capture_batch_dict(existing) + + result = conn.execute( + insert(capture_batches).values( + batch_key=batch_key, + source=source, + captured_at=captured_at, + item_count=len(items), + status="received", + payload_json=_json(payload), + results_json=_json([]), + created_at=_now(), + ingested_at=None, + ) + ) + batch_id = result.inserted_primary_key[0] + + processed: list[dict[str, Any]] = [] + for item in items: + kind = item.get("kind") + if kind == "resident": + self._insert_resident_row( + conn, + resident_key=item["resident_key"], + geography=item.get("geography", ""), + interests=item.get("interests", []), + lived_experience=item.get("lived_experience", []), + languages=item.get("languages", []), + ) + processed.append({"kind": "resident", "resident_key": item["resident_key"]}) + elif kind == "comment": + comment_id = self._insert_comment_row( + conn, + proposal_id=int(item["proposal_id"]), + text=item["text"], + channel=item.get("channel", source), + resident_key=item.get("resident_key"), + geography=item.get("geography", ""), + metadata={**item.get("metadata", {}), "batch_key": batch_key, "source": source}, + ) + processed.append({"kind": "comment", "comment_id": comment_id, "proposal_id": int(item["proposal_id"])}) + elif kind == "preference": + preference_id = self._insert_preference_row( + conn, + proposal_id=int(item["proposal_id"]), + resident_key=item["resident_key"], + score=float(item["score"]), + channel=item.get("channel", source), + ) + processed.append({"kind": "preference", "preference_id": preference_id, "proposal_id": int(item["proposal_id"])}) + elif kind == "ledger": + ledger_id = self._insert_ledger_row( + conn, + proposal_id=int(item["proposal_id"]), + kind=item["entry_kind"], + payload={**item.get("payload", {}), "batch_key": batch_key, "source": source}, + ) + processed.append({"kind": "ledger", "ledger_id": ledger_id, "proposal_id": int(item["proposal_id"])}) + elif kind == "proposal_version": + version_info = self._record_proposal_version_row( + conn, + proposal_id=int(item["proposal_id"]), + rationale=item.get("rationale", ""), + objections=item.get("objections", []), + resolution_status=item.get("resolution_status", "open"), + ) + processed.append({"kind": "proposal_version", **version_info}) + else: + raise ValueError(f"unsupported capture kind: {kind!r}") + + conn.execute( + update(capture_batches) + .where(capture_batches.c.id == batch_id) + .values(status="processed", results_json=_json(processed), ingested_at=_now()) + ) + + return self.get_capture_batch(batch_key) + + def get_capture_batch(self, batch_key: str) -> dict[str, Any]: + with self.db.engine.begin() as conn: + row = conn.execute(select(capture_batches).where(capture_batches.c.batch_key == batch_key)).mappings().first() + if row is None: + raise KeyError(f"capture batch {batch_key} not found") + return self._capture_batch_dict(row) + def aggregate_preferences(self, proposal_id: int, epsilon: float | None = None, seed: int = 0) -> dict[str, Any]: with self.db.engine.begin() as conn: rows = conn.execute(select(preferences.c.score).where(preferences.c.proposal_id == proposal_id)).all() @@ -302,46 +359,13 @@ class CivicSwarmService: def record_ledger_entry(self, proposal_id: int, kind: str, payload: dict[str, Any]) -> dict[str, Any]: with self.db.engine.begin() as conn: - result = conn.execute( - insert(ledger_entries).values( - proposal_id=proposal_id, - kind=kind, - payload_json=_json(payload), - created_at=_now(), - ) - ) - return {"id": result.inserted_primary_key[0], "proposal_id": proposal_id, "kind": kind, "payload": payload} + result_id = self._insert_ledger_row(conn, proposal_id, kind, payload) + return {"id": result_id, "proposal_id": proposal_id, "kind": kind, "payload": payload} def record_proposal_version(self, proposal_id: int, rationale: str, objections: list[str], resolution_status: str) -> dict[str, Any]: with self.db.engine.begin() as conn: - current = conn.execute( - select(func.max(proposal_versions.c.version)).where(proposal_versions.c.proposal_id == proposal_id) - ).scalar_one_or_none() - version = int(current or 0) + 1 - result = conn.execute( - insert(proposal_versions).values( - proposal_id=proposal_id, - version=version, - rationale=rationale, - objections_json=_json(objections), - resolution_status=resolution_status, - created_at=_now(), - ) - ) - conn.execute( - update(proposals) - .where(proposals.c.id == proposal_id) - .values(version=version, updated_at=_now(), status=resolution_status) - ) - conn.execute( - insert(ledger_entries).values( - proposal_id=proposal_id, - kind="proposal_version_recorded", - payload_json=_json({"version": version, "resolution_status": resolution_status}), - created_at=_now(), - ) - ) - return {"id": result.inserted_primary_key[0], "proposal_id": proposal_id, "version": version, "resolution_status": resolution_status} + version_info = self._record_proposal_version_row(conn, proposal_id, rationale, objections, resolution_status) + return version_info def list_proposal_versions(self, proposal_id: int) -> list[dict[str, Any]]: with self.db.engine.begin() as conn: @@ -466,6 +490,139 @@ class CivicSwarmService: overlap = len(left_tokens & right_tokens) return overlap / math.sqrt(len(left_tokens) * len(right_tokens)) + def _insert_resident_row( + self, + conn, + resident_key: str, + geography: str = "", + interests: list[str] | None = None, + lived_experience: list[str] | None = None, + languages: list[str] | None = None, + ) -> str: + interests = interests or [] + lived_experience = lived_experience or [] + languages = languages or [] + existing = conn.execute(select(residents).where(residents.c.resident_key == resident_key)).mappings().first() + values = dict( + resident_key=resident_key, + geography=geography, + interests_json=_json(interests), + lived_experience_json=_json(lived_experience), + languages_json=_json(languages), + updated_at=_now(), + ) + if existing is None: + values["created_at"] = _now() + conn.execute(insert(residents).values(**values)) + else: + conn.execute(update(residents).where(residents.c.resident_key == resident_key).values(**values)) + return resident_key + + def _insert_comment_row( + self, + conn, + proposal_id: int, + text: str, + channel: str, + resident_key: str | None, + geography: str, + metadata: dict[str, Any], + ) -> int: + language = detect_language(text) + result = conn.execute( + insert(comments).values( + proposal_id=proposal_id, + resident_key=self._token(resident_key), + channel=channel, + geography=geography, + language=language, + text=text, + metadata_json=_json(metadata), + created_at=_now(), + ) + ) + comment_id = result.inserted_primary_key[0] + self._insert_ledger_row( + conn, + proposal_id, + "comment_received", + {"comment_id": comment_id, "channel": channel, "language": language}, + ) + return int(comment_id) + + def _insert_preference_row(self, conn, proposal_id: int, resident_key: str, score: float, channel: str) -> int: + result = conn.execute( + insert(preferences).values( + proposal_id=proposal_id, + resident_key=self._token(resident_key) or resident_key, + score=float(score), + channel=channel, + created_at=_now(), + ) + ) + preference_id = int(result.inserted_primary_key[0]) + self._insert_ledger_row( + conn, + proposal_id, + "preference_submitted", + {"preference_id": preference_id, "channel": channel}, + ) + return preference_id + + def _insert_ledger_row(self, conn, proposal_id: int, kind: str, payload: dict[str, Any]) -> int: + result = conn.execute( + insert(ledger_entries).values( + proposal_id=proposal_id, + kind=kind, + payload_json=_json(payload), + created_at=_now(), + ) + ) + return int(result.inserted_primary_key[0]) + + def _record_proposal_version_row( + self, + conn, + proposal_id: int, + rationale: str, + objections: list[str], + resolution_status: str, + ) -> dict[str, Any]: + current = conn.execute(select(func.max(proposal_versions.c.version)).where(proposal_versions.c.proposal_id == proposal_id)).scalar_one_or_none() + version = int(current or 0) + 1 + result = conn.execute( + insert(proposal_versions).values( + proposal_id=proposal_id, + version=version, + rationale=rationale, + objections_json=_json(objections), + resolution_status=resolution_status, + created_at=_now(), + ) + ) + conn.execute(update(proposals).where(proposals.c.id == proposal_id).values(version=version, updated_at=_now(), status=resolution_status)) + self._insert_ledger_row( + conn, + proposal_id, + "proposal_version_recorded", + {"version": version, "resolution_status": resolution_status}, + ) + return {"id": int(result.inserted_primary_key[0]), "proposal_id": proposal_id, "version": version, "resolution_status": resolution_status} + + def _capture_batch_dict(self, row: dict[str, Any]) -> dict[str, Any]: + return { + "id": row["id"], + "batch_key": row["batch_key"], + "source": row["source"], + "captured_at": row["captured_at"].isoformat(), + "item_count": row["item_count"], + "status": row["status"], + "payload": _parse_json(row["payload_json"], {}), + "results": _parse_json(row["results_json"], []), + "created_at": row["created_at"].isoformat(), + "ingested_at": row["ingested_at"].isoformat() if row["ingested_at"] else None, + } + def _proposal_dict(self, row: dict[str, Any]) -> dict[str, Any]: return { "id": row["id"], diff --git a/tests/test_service.py b/tests/test_service.py index 163a4f0..aa41b07 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -2,6 +2,8 @@ from __future__ import annotations from pathlib import Path +from fastapi.testclient import TestClient + from civicswarm.api import create_app from civicswarm.service import CivicSwarmService @@ -89,3 +91,70 @@ def test_private_preference_aggregation_supports_noise(tmp_path: Path): assert clean["count"] == 3 assert noisy["count"] >= 1 assert noisy["mean_score"] != clean["mean_score"] or noisy["count"] != clean["count"] + + +def test_capture_batch_ingest_is_atomic_and_idempotent(tmp_path: Path): + service = build_service(tmp_path) + proposal_id = seed_service(service) + + batch = service.ingest_capture_batch( + "ward-3-mobile-001", + "sms", + [ + {"kind": "resident", "resident_key": "resident-d", "geography": "ward-3", "interests": ["schools"], "languages": ["en"]}, + {"kind": "comment", "proposal_id": proposal_id, "text": "Please keep crossings safe for elders.", "resident_key": "resident-d", "channel": "sms"}, + {"kind": "preference", "proposal_id": proposal_id, "resident_key": "resident-d", "score": 0.8, "channel": "sms"}, + {"kind": "ledger", "proposal_id": proposal_id, "entry_kind": "facilitator_note", "payload": {"note": "captured at block party"}}, + ], + ) + + assert batch["batch_key"] == "ward-3-mobile-001" + assert batch["status"] == "processed" + assert len(batch["results"]) == 4 + assert service.get_resident("resident-d")["geography"] == "ward-3" + comments = service.list_comments(proposal_id) + assert any(comment["metadata"].get("batch_key") == "ward-3-mobile-001" for comment in comments) + assert service.aggregate_preferences(proposal_id)["count"] == 4 + + duplicate = service.ingest_capture_batch( + "ward-3-mobile-001", + "sms", + [], + ) + assert duplicate["id"] == batch["id"] + assert duplicate["results"] == batch["results"] + assert len(service.list_comments(proposal_id)) == len(comments) + + +def test_capture_batch_endpoint_surfaces_audit_payload(tmp_path: Path): + app = create_app(tmp_path / "batch-api.sqlite") + client = TestClient(app) + + proposal = client.post( + "/proposals", + json={ + "city": "Springfield", + "title": "Crosswalk renewal", + "body": "Renew the busiest crosswalks near the library.", + "geography": "ward-4", + "tags": ["safety", "mobility"], + }, + ).json() + + response = client.post( + "/capture-batches", + json={ + "batch_key": "ward-4-townhall-01", + "source": "field", + "items": [ + {"kind": "comment", "proposal_id": proposal["id"], "text": "I support the renewal.", "channel": "field"}, + ], + }, + ) + + assert response.status_code == 200 + payload = response.json() + assert payload["status"] == "processed" + assert payload["results"][0]["kind"] == "comment" + fetched = client.get("/capture-batches/ward-4-townhall-01").json() + assert fetched["batch_key"] == "ward-4-townhall-01"