663 lines
28 KiB
Python
663 lines
28 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import math
|
|
import re
|
|
from collections import defaultdict
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Iterable
|
|
|
|
import numpy as np
|
|
from sqlalchemy import delete, func, insert, select, update
|
|
|
|
from .analysis import (
|
|
cluster_texts,
|
|
detect_language,
|
|
laplace_noise,
|
|
sentiment_score,
|
|
summarize_multilingual_comments,
|
|
tokenize,
|
|
)
|
|
from .db import (
|
|
capture_batches,
|
|
comments,
|
|
Database,
|
|
create_database,
|
|
ledger_entries,
|
|
preferences,
|
|
proposal_versions,
|
|
proposals,
|
|
residents,
|
|
routes,
|
|
sqlite_path_url,
|
|
)
|
|
|
|
|
|
def _json(value: Any) -> str:
|
|
return json.dumps(value, ensure_ascii=False, sort_keys=True)
|
|
|
|
|
|
def _parse_json(value: str | None, default: Any) -> Any:
|
|
if not value:
|
|
return default
|
|
return json.loads(value)
|
|
|
|
|
|
def _now() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
def _normalize_terms(values: Iterable[str]) -> set[str]:
|
|
terms: set[str] = set()
|
|
for value in values:
|
|
terms.update(tokenize(value))
|
|
return terms
|
|
|
|
|
|
@dataclass
|
|
class CivicSwarmService:
|
|
db_url: str
|
|
secret: str = "civicswarm"
|
|
db: Database | None = None
|
|
|
|
def __post_init__(self) -> None:
|
|
self.db = create_database(self.db_url)
|
|
|
|
@classmethod
|
|
def from_path(cls, path: str | Path, secret: str = "civicswarm") -> "CivicSwarmService":
|
|
return cls(db_url=sqlite_path_url(path), secret=secret)
|
|
|
|
def _token(self, resident_key: str | None) -> str | None:
|
|
if resident_key is None:
|
|
return None
|
|
return hmac.new(self.secret.encode(), resident_key.encode(), hashlib.sha256).hexdigest()
|
|
|
|
def create_proposal(self, city: str, title: str, body: str, geography: str = "", tags: list[str] | None = None) -> dict[str, Any]:
|
|
tags = tags or []
|
|
with self.db.engine.begin() as conn:
|
|
result = conn.execute(
|
|
insert(proposals).values(
|
|
city=city,
|
|
title=title,
|
|
body=body,
|
|
geography=geography,
|
|
tags_json=_json(tags),
|
|
status="draft",
|
|
version=1,
|
|
created_at=_now(),
|
|
updated_at=_now(),
|
|
)
|
|
)
|
|
proposal_id = result.inserted_primary_key[0]
|
|
conn.execute(
|
|
insert(ledger_entries).values(
|
|
proposal_id=proposal_id,
|
|
kind="proposal_created",
|
|
payload_json=_json({"title": title, "tags": tags, "geography": geography}),
|
|
created_at=_now(),
|
|
)
|
|
)
|
|
conn.execute(
|
|
insert(proposal_versions).values(
|
|
proposal_id=proposal_id,
|
|
version=1,
|
|
rationale="initial proposal",
|
|
objections_json=_json([]),
|
|
resolution_status="open",
|
|
created_at=_now(),
|
|
)
|
|
)
|
|
return self.get_proposal(proposal_id)
|
|
|
|
def get_proposal(self, proposal_id: int) -> dict[str, Any]:
|
|
with self.db.engine.begin() as conn:
|
|
row = conn.execute(select(proposals).where(proposals.c.id == proposal_id)).mappings().first()
|
|
if row is None:
|
|
raise KeyError(f"proposal {proposal_id} not found")
|
|
return self._proposal_dict(row)
|
|
|
|
def list_comments(self, proposal_id: int) -> list[dict[str, Any]]:
|
|
with self.db.engine.begin() as conn:
|
|
rows = conn.execute(select(comments).where(comments.c.proposal_id == proposal_id).order_by(comments.c.id.asc())).mappings().all()
|
|
return [self._comment_dict(row) for row in rows]
|
|
|
|
def register_resident(
|
|
self,
|
|
resident_key: str,
|
|
geography: str = "",
|
|
interests: list[str] | None = None,
|
|
lived_experience: list[str] | None = None,
|
|
languages: list[str] | None = None,
|
|
) -> dict[str, Any]:
|
|
interests = interests or []
|
|
lived_experience = lived_experience or []
|
|
languages = languages or []
|
|
with self.db.engine.begin() as conn:
|
|
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 self.get_resident(resident_key)
|
|
|
|
def get_resident(self, resident_key: str) -> dict[str, Any]:
|
|
with self.db.engine.begin() as conn:
|
|
row = conn.execute(select(residents).where(residents.c.resident_key == resident_key)).mappings().first()
|
|
if row is None:
|
|
raise KeyError(f"resident {resident_key} not found")
|
|
return self._resident_dict(row)
|
|
|
|
def add_comment(
|
|
self,
|
|
proposal_id: int,
|
|
text: str,
|
|
channel: str = "mobile",
|
|
resident_key: str | None = None,
|
|
geography: str = "",
|
|
metadata: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
metadata = metadata or {}
|
|
with self.db.engine.begin() as conn:
|
|
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]:
|
|
with self.db.engine.begin() as conn:
|
|
row = conn.execute(select(comments).where(comments.c.id == comment_id)).mappings().first()
|
|
if row is None:
|
|
raise KeyError(f"comment {comment_id} not found")
|
|
return self._comment_dict(row)
|
|
|
|
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:
|
|
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()
|
|
scores = [float(row[0]) for row in rows]
|
|
count = len(scores)
|
|
total = float(sum(scores))
|
|
mean = total / count if count else 0.0
|
|
if epsilon and epsilon > 0:
|
|
scale = 1.0 / float(epsilon)
|
|
total += laplace_noise(scale, seed=seed)
|
|
count = max(1, int(round(count + laplace_noise(scale / 2.0, seed=seed + 1))))
|
|
mean = total / count
|
|
return {"proposal_id": proposal_id, "count": count, "mean_score": mean, "raw_total": float(sum(scores))}
|
|
|
|
def cluster_comments(self, proposal_id: int) -> list[dict[str, Any]]:
|
|
comments_data = self.list_comments(proposal_id)
|
|
return cluster_texts([comment["text"] for comment in comments_data])
|
|
|
|
def summarize_comments(self, proposal_id: int, max_sentences: int = 3) -> dict[str, list[str]]:
|
|
comments_data = self.list_comments(proposal_id)
|
|
return summarize_multilingual_comments(comments_data, max_sentences=max_sentences)
|
|
|
|
def route_proposal(self, proposal_id: int, top_n: int = 3) -> dict[str, Any]:
|
|
proposal = self.get_proposal(proposal_id)
|
|
fragments = self._proposal_fragments(proposal["body"])
|
|
with self.db.engine.begin() as conn:
|
|
resident_rows = conn.execute(select(residents)).mappings().all()
|
|
|
|
routed_fragments = []
|
|
for index, fragment in enumerate(fragments):
|
|
ranked = self._rank_residents_for_fragment(proposal, fragment, resident_rows)
|
|
top_matches = ranked[:top_n]
|
|
routed_fragments.append({"fragment_index": index, "fragment": fragment, "matches": top_matches})
|
|
with self.db.engine.begin() as conn:
|
|
conn.execute(delete(routes).where(routes.c.proposal_id == proposal_id, routes.c.fragment_index == index))
|
|
for match in top_matches:
|
|
conn.execute(
|
|
insert(routes).values(
|
|
proposal_id=proposal_id,
|
|
fragment_index=index,
|
|
resident_key=match["resident_key"],
|
|
score=match["score"],
|
|
rationale_json=_json(match["reasons"]),
|
|
created_at=_now(),
|
|
)
|
|
)
|
|
return {"proposal_id": proposal_id, "fragments": routed_fragments}
|
|
|
|
def build_dashboard(self, proposal_id: int) -> dict[str, Any]:
|
|
proposal = self.get_proposal(proposal_id)
|
|
comments_data = self.list_comments(proposal_id)
|
|
summaries = self.summarize_comments(proposal_id)
|
|
preferences_summary = self.aggregate_preferences(proposal_id)
|
|
routes_summary = self.route_proposal(proposal_id)
|
|
|
|
consensus_pockets = []
|
|
unresolved_conflicts = []
|
|
outreach_gaps = []
|
|
|
|
for fragment in routes_summary["fragments"]:
|
|
fragment_text = fragment["fragment"]
|
|
fragment_comments = [comment for comment in comments_data if any(term in comment["text"].lower() for term in tokenize(fragment_text))]
|
|
sentiment_values = [sentiment_score(comment["text"]) for comment in fragment_comments]
|
|
if sentiment_values and np.mean(sentiment_values) > 0:
|
|
consensus_pockets.append({"fragment_index": fragment["fragment_index"], "fragment": fragment_text, "support": float(np.mean(sentiment_values))})
|
|
if sentiment_values and min(sentiment_values) < 0 < max(sentiment_values):
|
|
unresolved_conflicts.append({"fragment_index": fragment["fragment_index"], "fragment": fragment_text, "support": float(np.mean(sentiment_values)), "spread": float(max(sentiment_values) - min(sentiment_values))})
|
|
if not fragment["matches"]:
|
|
outreach_gaps.append({"fragment_index": fragment["fragment_index"], "fragment": fragment_text})
|
|
|
|
return {
|
|
"proposal": proposal,
|
|
"summaries": summaries,
|
|
"preference_aggregate": preferences_summary,
|
|
"consensus_pockets": consensus_pockets,
|
|
"unresolved_conflicts": unresolved_conflicts,
|
|
"outreach_gaps": outreach_gaps,
|
|
}
|
|
|
|
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_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:
|
|
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:
|
|
rows = conn.execute(
|
|
select(proposal_versions).where(proposal_versions.c.proposal_id == proposal_id).order_by(proposal_versions.c.version.asc())
|
|
).mappings().all()
|
|
return [
|
|
{
|
|
"id": row["id"],
|
|
"proposal_id": row["proposal_id"],
|
|
"version": row["version"],
|
|
"rationale": row["rationale"],
|
|
"objections": _parse_json(row["objections_json"], []),
|
|
"resolution_status": row["resolution_status"],
|
|
"created_at": row["created_at"].isoformat(),
|
|
}
|
|
for row in rows
|
|
]
|
|
|
|
def list_ledger(self, proposal_id: int) -> list[dict[str, Any]]:
|
|
with self.db.engine.begin() as conn:
|
|
rows = conn.execute(select(ledger_entries).where(ledger_entries.c.proposal_id == proposal_id).order_by(ledger_entries.c.id.asc())).mappings().all()
|
|
return [
|
|
{
|
|
"id": row["id"],
|
|
"proposal_id": row["proposal_id"],
|
|
"kind": row["kind"],
|
|
"payload": _parse_json(row["payload_json"], {}),
|
|
"created_at": row["created_at"].isoformat(),
|
|
}
|
|
for row in rows
|
|
]
|
|
|
|
def export_brief(self, proposal_id: int) -> dict[str, Any]:
|
|
proposal = self.get_proposal(proposal_id)
|
|
comments_data = self.list_comments(proposal_id)
|
|
dashboard = self.build_dashboard(proposal_id)
|
|
ledger = self.list_ledger(proposal_id)
|
|
clusters = self.cluster_comments(proposal_id)
|
|
versions = self.list_proposal_versions(proposal_id)
|
|
|
|
provenance = [
|
|
{"comment_id": comment["id"], "language": comment["language"], "channel": comment["channel"]}
|
|
for comment in comments_data
|
|
]
|
|
return {
|
|
"proposal": proposal,
|
|
"headline": proposal["title"],
|
|
"summary": dashboard["summaries"],
|
|
"topic_clusters": clusters,
|
|
"preference_aggregate": dashboard["preference_aggregate"],
|
|
"consensus_pockets": dashboard["consensus_pockets"],
|
|
"unresolved_conflicts": dashboard["unresolved_conflicts"],
|
|
"outreach_gaps": dashboard["outreach_gaps"],
|
|
"ledger": ledger,
|
|
"proposal_versions": versions,
|
|
"provenance": provenance,
|
|
}
|
|
|
|
def _proposal_fragments(self, body: str) -> list[str]:
|
|
fragments = []
|
|
for chunk in body.split("\n"):
|
|
chunk = chunk.strip(" -\t")
|
|
if not chunk:
|
|
continue
|
|
fragments.extend([part.strip() for part in re.split(r"(?<=[.!?])\s+", chunk) if part.strip()])
|
|
return fragments or ([body.strip()] if body.strip() else [])
|
|
|
|
def _rank_residents_for_fragment(self, proposal: dict[str, Any], fragment: str, resident_rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
proposal_tags = set(proposal.get("tags", []))
|
|
fragment_terms = _normalize_terms([fragment])
|
|
ranked = []
|
|
for resident in resident_rows:
|
|
interests = set(_parse_json(resident["interests_json"], []))
|
|
experience = set(_parse_json(resident["lived_experience_json"], []))
|
|
languages = set(_parse_json(resident["languages_json"], []))
|
|
resident_geo = (resident["geography"] or "").lower()
|
|
score = 0.0
|
|
reasons = []
|
|
|
|
if resident_geo and resident_geo in (proposal["geography"] or "").lower():
|
|
score += 0.25
|
|
reasons.append("geography_match")
|
|
|
|
overlap = len(_normalize_terms(proposal_tags) & interests)
|
|
if overlap:
|
|
score += min(0.3, 0.1 * overlap)
|
|
reasons.append("interest_overlap")
|
|
|
|
experience_overlap = len(fragment_terms & _normalize_terms(experience))
|
|
if experience_overlap:
|
|
score += min(0.25, 0.12 * experience_overlap)
|
|
reasons.append("lived_experience_match")
|
|
|
|
fragment_language = detect_language(fragment)
|
|
if fragment_language in languages:
|
|
score += 0.1
|
|
reasons.append("language_match")
|
|
|
|
profile_text = " ".join([resident["geography"], " ".join(sorted(interests)), " ".join(sorted(experience)), " ".join(sorted(languages))])
|
|
lexical = self._similarity(fragment, profile_text)
|
|
score += 0.35 * lexical
|
|
if lexical > 0:
|
|
reasons.append("lexical_similarity")
|
|
|
|
ranked.append(
|
|
{
|
|
"resident_key": resident["resident_key"],
|
|
"score": round(float(score), 4),
|
|
"reasons": reasons,
|
|
}
|
|
)
|
|
|
|
ranked.sort(key=lambda item: (-item["score"], item["resident_key"]))
|
|
return [item for item in ranked if item["score"] > 0]
|
|
|
|
def _similarity(self, left: str, right: str) -> float:
|
|
left_tokens = _normalize_terms([left])
|
|
right_tokens = _normalize_terms([right])
|
|
if not left_tokens or not right_tokens:
|
|
return 0.0
|
|
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"],
|
|
"city": row["city"],
|
|
"title": row["title"],
|
|
"body": row["body"],
|
|
"status": row["status"],
|
|
"geography": row["geography"],
|
|
"tags": _parse_json(row["tags_json"], []),
|
|
"version": row["version"],
|
|
"created_at": row["created_at"].isoformat(),
|
|
"updated_at": row["updated_at"].isoformat(),
|
|
}
|
|
|
|
def _resident_dict(self, row: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"resident_key": row["resident_key"],
|
|
"geography": row["geography"],
|
|
"interests": _parse_json(row["interests_json"], []),
|
|
"lived_experience": _parse_json(row["lived_experience_json"], []),
|
|
"languages": _parse_json(row["languages_json"], []),
|
|
"created_at": row["created_at"].isoformat(),
|
|
"updated_at": row["updated_at"].isoformat(),
|
|
}
|
|
|
|
def _comment_dict(self, row: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"id": row["id"],
|
|
"proposal_id": row["proposal_id"],
|
|
"resident_key": row["resident_key"],
|
|
"channel": row["channel"],
|
|
"geography": row["geography"],
|
|
"language": row["language"],
|
|
"text": row["text"],
|
|
"metadata": _parse_json(row["metadata_json"], {}),
|
|
"created_at": row["created_at"].isoformat(),
|
|
}
|