feat: persist course sync field diffs

Author: aryans-15

ferry/database/generate_changelog.py

@@ -2,6 +2,7 @@
 from pathlib import Path
 from time import gmtime, strftime
 from typing import cast, Any, TypedDict, Callable
+import numpy as np
 import networkx as nx
 
 
@@ -323,6 +324,171 @@ def register_junction_changes(
         )
 
 
+def build_course_id_to_changes(
+    diff: dict[str, DiffRecord],
+    tables_old: dict[str, pd.DataFrame],
+    tables: dict[str, pd.DataFrame],
+) -> dict[int, dict[str, tuple[Any, Any]]]:
+    """
+    Aggregate per-course field/junction changes the same way as the markdown changelog.
+    Used to persist structured diffs (e.g. for email alerts).
+    """
+    courses_old_indexed = tables_old["courses"].set_index("course_id")
+    courses_new_indexed = tables["courses"].set_index("course_id")
+
+    course_id_to_changes: dict[int, dict[str, tuple[Any, Any]]] = {}
+    for _, course in diff["courses"]["changed_rows"].iterrows():
+        course_id = cast(int, course["course_id"])
+        if course_id not in course_id_to_changes:
+            course_id_to_changes[course_id] = {}
+        for column in course["columns_changed"]:
+            if column in computed_columns["courses"]:
+                continue
+            old_val = courses_old_indexed.loc[course_id, column]
+            new_val = courses_new_indexed.loc[course_id, column]
+            course_id_to_changes[course_id][column] = (old_val, new_val)
+    register_junction_changes(
+        course_id_to_changes,
+        diff,
+        tables_old,
+        tables,
+        "course_professors",
+    )
+    register_junction_changes(
+        course_id_to_changes,
+        diff,
+        tables_old,
+        tables,
+        "course_flags",
+    )
+    register_junction_changes(
+        course_id_to_changes,
+        diff,
+        tables_old,
+        tables,
+        "listings",
+        ["course_id"],
+    )
+    return course_id_to_changes
+
+
+def _json_prepare_scalar(value: Any) -> Any:
+    if value is None:
+        return None
+    if isinstance(value, np.ndarray):
+        return _json_prepare_scalar(value.tolist())
+    if isinstance(value, (list, tuple)):
+        return [_json_prepare_scalar(v) for v in value]
+    if isinstance(value, dict):
+        return {str(k): _json_prepare_scalar(v) for k, v in value.items()}
+    if isinstance(value, (np.integer, np.int64, np.int32)):
+        return int(value)
+    if isinstance(value, (np.floating, np.float64, np.float32)):
+        if np.isnan(value):
+            return None
+        return float(value)
+    if isinstance(value, np.bool_):
+        return bool(value)
+    if isinstance(value, pd.Timestamp):
+        return value.isoformat()
+    if safe_scalar_na(value):
+        return None
+    if isinstance(value, (bytes, bytearray)):
+        return value.decode("utf-8", errors="replace")
+    return value
+
+
+def safe_scalar_na(value: Any) -> bool:
+    try:
+        return bool(pd.isna(value))
+    except (TypeError, ValueError):
+        return False
+
+
+def _listing_rows_for_json(df: pd.DataFrame) -> list[dict[str, Any]]:
+    if df.empty:
+        return []
+    cols = [c for c in ("listing_id", "season_code", "course_code", "section", "crn") if c in df.columns]
+    out: list[dict[str, Any]] = []
+    for rec in df[cols].replace({np.nan: None}).to_dict(orient="records"):
+        row = {k: _json_prepare_scalar(v) for k, v in rec.items()}
+        out.append(row)
+    return out
+
+
+def _serialize_change_pair(old: Any, new: Any, prof_info: pd.DataFrame, flag_info: pd.DataFrame) -> dict[str, Any]:
+    if isinstance(old, pd.DataFrame) and isinstance(new, pd.DataFrame):
+        if "professor_id" in old.columns and "professor_id" in new.columns:
+            old_ids = sorted(int(x) for x in old["professor_id"].tolist())
+            new_ids = sorted(int(x) for x in new["professor_id"].tolist())
+            return {
+                "old": {
+                    "professor_ids": old_ids,
+                    "names": prof_info.reindex(old_ids)["name"].fillna("").tolist(),
+                },
+                "new": {
+                    "professor_ids": new_ids,
+                    "names": prof_info.reindex(new_ids)["name"].fillna("").tolist(),
+                },
+            }
+        if "flag_id" in old.columns and "flag_id" in new.columns:
+            old_ids = sorted(int(x) for x in old["flag_id"].tolist())
+            new_ids = sorted(int(x) for x in new["flag_id"].tolist())
+            return {
+                "old": {
+                    "flag_ids": old_ids,
+                    "texts": flag_info.reindex(old_ids)["flag_text"].fillna("").tolist(),
+                },
+                "new": {
+                    "flag_ids": new_ids,
+                    "texts": flag_info.reindex(new_ids)["flag_text"].fillna("").tolist(),
+                },
+            }
+        if "listing_id" in old.columns and "listing_id" in new.columns:
+            old_rows = sorted(
+                _listing_rows_for_json(old), key=lambda r: r.get("listing_id") or 0
+            )
+            new_rows = sorted(
+                _listing_rows_for_json(new), key=lambda r: r.get("listing_id") or 0
+            )
+            return {"old": old_rows, "new": new_rows}
+    return {"old": _json_prepare_scalar(old), "new": _json_prepare_scalar(new)}
+
+
+def serialize_course_id_to_changes(
+    course_id_to_changes: dict[int, dict[str, tuple[Any, Any]]],
+    tables_old: dict[str, pd.DataFrame],
+    tables: dict[str, pd.DataFrame],
+) -> dict[int, dict[str, Any]]:
+    """
+    Convert build_course_id_to_changes output to JSON-serializable dicts per course_id.
+    Omits courses with no remaining keys after serialization (e.g. empty junction pairs).
+    """
+    prof_info = (
+        tables["professors"]
+        .set_index("professor_id")
+        .combine_first(tables_old["professors"].set_index("professor_id"))
+    )
+    flag_info = (
+        tables["flags"]
+        .set_index("flag_id")
+        .combine_first(tables_old["flags"].set_index("flag_id"))
+    )
+    result: dict[int, dict[str, Any]] = {}
+    for course_id, changes in course_id_to_changes.items():
+        if not changes:
+            continue
+        serialized_fields: dict[str, Any] = {}
+        for field, (old, new) in changes.items():
+            pair = _serialize_change_pair(old, new, prof_info, flag_info)
+            if pair["old"] == pair["new"]:
+                continue
+            serialized_fields[field] = pair
+        if serialized_fields:
+            result[course_id] = serialized_fields
+    return result
+
+
 def print_courses_diff(
     diff: dict[str, DiffRecord],
     tables_old: dict[str, pd.DataFrame],
@@ -396,45 +562,9 @@ def print_courses_diff(
 
     # Process changed courses with indexed lookups
     course_updates = ""
-    course_id_to_changes: dict[int, dict[str, tuple[Any, Any]]] = {}
-    for _, course in diff["courses"]["changed_rows"].iterrows():
-        course_id = cast(int, course["course_id"])
-        if course_id not in course_id_to_changes:
-            course_id_to_changes[course_id] = {}
-        for column in course["columns_changed"]:
-            if column in computed_columns["courses"]:
-                continue
-            # Use indexed lookup instead of filtering
-            old_val = courses_old_indexed.loc[course_id, column]
-            new_val = courses_new_indexed.loc[course_id, column]
-            course_id_to_changes[course_id][column] = (old_val, new_val)
-    register_junction_changes(
-        course_id_to_changes,
-        diff,
-        tables_old,
-        tables,
-        "course_professors",
-    )
-    register_junction_changes(
-        course_id_to_changes,
-        diff,
-        tables_old,
-        tables,
-        "course_flags",
-    )
+    course_id_to_changes = build_course_id_to_changes(diff, tables_old, tables)
     # Note: course_meetings uses UPSERT logic, so junction change tracking is skipped
     # Meeting changes (if any) will not be reflected accurately in the changelog
-    
-    register_junction_changes(
-        course_id_to_changes,
-        diff,
-        tables_old,
-        tables,
-        "listings",
-        # Only include changes to the course_id column, because changes to
-        # other columns are reported by the listings section
-        ["course_id"],
-    )
     prof_info = (
         tables["professors"]
         .set_index("professor_id")

ferry/database/models.py

@@ -305,6 +305,12 @@ class Course(BaseModel):
         comment="[computed] Whether last enrollment offering is with same professor as current.",
     )
 
+    last_sync_diff = Column(
+        JSONB,
+        comment="Per-field old/new values from the most recent Ferry sync for this course (user-visible changes only). Null when nothing relevant changed in the last run.",
+        nullable=True,
+    )
+
 
 class Listing(BaseModel):
     """

ferry/database/sync_db_courses.py

@@ -9,7 +9,14 @@
 from psycopg2.extensions import register_adapter, AsIs
 
 from ferry.database import Database
-from .generate_changelog import print_diff, computed_columns, primary_keys, DiffRecord
+from .generate_changelog import (
+    print_diff,
+    computed_columns,
+    primary_keys,
+    DiffRecord,
+    build_course_id_to_changes,
+    serialize_course_id_to_changes,
+)
 
 
 register_adapter(np.int64, AsIs)
@@ -47,7 +54,8 @@ def get_tables_from_db(database_connect_string: str) -> dict[str, pd.DataFrame]:
 
     return {
         table_name: pd.read_sql_table(table_name, con=conn).drop(
-            columns=["time_added", "last_updated"], errors="ignore"
+            columns=["time_added", "last_updated", "last_sync_diff"],
+            errors="ignore",
         )
         for table_name in primary_keys.keys()
     }
@@ -701,6 +709,12 @@ def sync_db_courses(
 
     diff = generate_diff(tables_old_for_diff, tables_for_diff)
 
+    course_sync_diff_by_id = serialize_course_id_to_changes(
+        build_course_id_to_changes(diff, tables_old, tables),
+        tables_old,
+        tables,
+    )
+
     print_diff(diff, tables_old, tables, data_dir / "change_log")
 
     inspector = inspect(db.Engine)
@@ -726,6 +740,10 @@ def sync_db_courses(
                     )
                 )
 
+        course_col_names = {c["name"] for c in inspector.get_columns("courses")}
+        if "last_sync_diff" not in course_col_names:
+            conn.execute(text("ALTER TABLE courses ADD COLUMN last_sync_diff JSONB"))
+
         # Process tables in dependency order (buildings before locations)
         location_mapping = {}
         for table_name in tables_order_add:
@@ -812,6 +830,34 @@ def sync_db_courses(
             commit_deletions(table_name, deleted_rows, conn)
         print("\033[F", end="")
 
+        changed_course_ids = list(course_sync_diff_by_id.keys())
+        if changed_course_ids:
+            ids_sql = ",".join(str(int(cid)) for cid in changed_course_ids)
+            conn.execute(
+                text(
+                    f"UPDATE courses SET last_sync_diff = NULL "
+                    f"WHERE last_sync_diff IS NOT NULL AND course_id NOT IN ({ids_sql})"
+                )
+            )
+        else:
+            conn.execute(
+                text(
+                    "UPDATE courses SET last_sync_diff = NULL WHERE last_sync_diff IS NOT NULL"
+                )
+            )
+
+        upd = text(
+            "UPDATE courses SET last_sync_diff = CAST(:payload AS jsonb) WHERE course_id = :course_id"
+        )
+        for course_id, payload in course_sync_diff_by_id.items():
+            conn.execute(
+                upd,
+                {
+                    "course_id": int(course_id),
+                    "payload": ujson.dumps(payload),
+                },
+            )
+
     # Print row counts for each table.
     print("\n[Table Statistics]")
     with db.Engine.begin() as conn: