feat: add delivery status tracking and crowd quality levels

data/crowd_db/cli.py

@@ -31,6 +31,18 @@ def _normalize_source_type(raw_source_type: str) -> dict[str, str]:
     }
 
 
+def _normalize_quality(confidence: Any) -> tuple[str, str]:
+    try:
+        numeric = float(confidence)
+    except (TypeError, ValueError):
+        return ("unknown", "未知")
+    if numeric >= 0.8:
+        return ("high", "A级（高置信）")
+    if numeric >= 0.5:
+        return ("usable", "B级（可用）")
+    return ("skeleton", "C级（骨架）")
+
+
 def _build_match(
     *,
     province: str,
@@ -40,6 +52,7 @@ def _build_match(
 ) -> dict[str, Any]:
     score_bounds = score_range.get("range") or [None, None]
     normalized = _normalize_source_type(str(provenance.get("source_type") or "derived"))
+    quality_level, quality_label = _normalize_quality(provenance.get("confidence"))
     return {
         "province": province,
         "school": recommendation.get("name", ""),
@@ -58,6 +71,8 @@ def _build_match(
         "source_type_label": normalized["source_type_label"],
         "source_type_icon": normalized["source_type_icon"],
         "confidence": provenance.get("confidence"),
+        "quality_level": quality_level,
+        "quality_label": quality_label,
         "last_updated": provenance.get("last_updated", ""),
     }
 
@@ -138,6 +153,7 @@ def _emit_human(payload: dict[str, Any]) -> None:
         print(f"source: {match['source']}")
         print(f"source_url: {match['source_url']}")
         print(f"confidence: {match['confidence']}")
+        print(f"quality_level: {match['quality_level']} ({match['quality_label']})")
         print(f"last_updated: {match['last_updated']}")
 
 

data/crowd_db/risk_report.py

@@ -108,6 +108,18 @@ def _normalize_provenance(metadata: Optional[Dict[str, Any]]) -> Dict[str, Any]:
         confidence = float(confidence) if confidence is not None else None
     except (TypeError, ValueError):
         confidence = None
+    if confidence is None:
+        quality_level = "unknown"
+        quality_label = "未知"
+    elif confidence >= 0.8:
+        quality_level = "high"
+        quality_label = "A级（高置信）"
+    elif confidence >= 0.5:
+        quality_level = "usable"
+        quality_label = "B级（可用）"
+    else:
+        quality_level = "skeleton"
+        quality_label = "C级（骨架）"
     data_year = metadata.get("data_year")
     try:
         data_year = int(data_year) if data_year is not None else None
@@ -122,6 +134,8 @@ def _normalize_provenance(metadata: Optional[Dict[str, Any]]) -> Dict[str, Any]:
         "source": metadata.get("source", ""),
         "source_url": metadata.get("source_url", ""),
         "confidence": confidence,
+        "quality_level": quality_level,
+        "quality_label": quality_label,
         "last_updated": metadata.get("last_updated", ""),
         "data_year": data_year,
     }

data/crowd_db/tests/test_risk_report.py

@@ -156,6 +156,8 @@ def test_risk_dict_includes_provenance_fields():
         "source",
         "source_url",
         "confidence",
+        "quality_level",
+        "quality_label",
         "last_updated",
         "data_year",
     ):
@@ -168,6 +170,8 @@ def test_risk_dict_includes_provenance_fields():
     assert r["last_updated"] == "2026-06-12"
     assert r["data_year"] == 2025
     assert 0 <= r["confidence"] <= 1
+    assert r["quality_level"] == "high"
+    assert r["quality_label"] == "A级（高置信）"
 
 
 def test_alternatives_remapped_to_school_field():

data/crowd_db/tests/test_trace_cli.py

@@ -54,6 +54,7 @@ def test_trace_cli_human_output_contains_required_lines(
     assert "湖南 / 2025年数据 / 长沙理工大学 / 会计学" in captured.out
     assert "source_url: https://" in captured.out
     assert "confidence: 0.85" in captured.out
+    assert "quality_level: high (A级（高置信）)" in captured.out
 
 
 def test_trace_cli_missing_school_returns_nonzero(

data/notifications/email_service.py

@@ -15,6 +15,10 @@ CREATE TABLE IF NOT EXISTS delivery_notifications (
     event_type TEXT NOT NULL,
     channel TEXT NOT NULL,
     payload_json TEXT NOT NULL,
+    status TEXT NOT NULL DEFAULT 'ready',
+    attempt_count INTEGER NOT NULL DEFAULT 1,
+    last_attempt_at TEXT,
+    failure_reason TEXT,
     created_at TEXT NOT NULL,
     UNIQUE(order_id, event_type),
     FOREIGN KEY (order_id) REFERENCES orders(id) ON DELETE CASCADE
@@ -28,6 +32,10 @@ class DeliveryNotificationEvent:
     event_type: str
     channel: str
     payload_json: str
+    status: str
+    attempt_count: int
+    last_attempt_at: str | None
+    failure_reason: str | None
     created_at: str
 
 
@@ -43,6 +51,7 @@ class DeliveryNotificationService:
         conn = apply_schema(db_path)
         conn.row_factory = sqlite3.Row
         conn.executescript(SCHEMA_SQL)
+        cls._ensure_columns(conn)
         conn.commit()
         return cls(conn)
 
@@ -50,8 +59,34 @@ class DeliveryNotificationService:
     def from_connection(cls, conn: sqlite3.Connection) -> "DeliveryNotificationService":
         conn.row_factory = sqlite3.Row
         conn.executescript(SCHEMA_SQL)
+        cls._ensure_columns(conn)
         return cls(conn, owns_connection=False)
 
+    @staticmethod
+    def _ensure_columns(conn: sqlite3.Connection) -> None:
+        columns = {
+            row[1]
+            for row in conn.execute(
+                "PRAGMA table_info(delivery_notifications)"
+            ).fetchall()
+        }
+        if "status" not in columns:
+            conn.execute(
+                "ALTER TABLE delivery_notifications ADD COLUMN status TEXT NOT NULL DEFAULT 'ready'"
+            )
+        if "attempt_count" not in columns:
+            conn.execute(
+                "ALTER TABLE delivery_notifications ADD COLUMN attempt_count INTEGER NOT NULL DEFAULT 1"
+            )
+        if "last_attempt_at" not in columns:
+            conn.execute(
+                "ALTER TABLE delivery_notifications ADD COLUMN last_attempt_at TEXT"
+            )
+        if "failure_reason" not in columns:
+            conn.execute(
+                "ALTER TABLE delivery_notifications ADD COLUMN failure_reason TEXT"
+            )
+
     def close(self) -> None:
         if self._owns_connection:
             self._conn.close()
@@ -61,16 +96,36 @@ class DeliveryNotificationService:
     ) -> None:
         try:
             self._conn.execute(
-                "INSERT INTO delivery_notifications(order_id, event_type, channel, payload_json, created_at) VALUES (?, 'report_ready', ?, ?, ?)",
-                (order_id, channel, payload_json, utc_now_iso()),
+                "INSERT INTO delivery_notifications(order_id, event_type, channel, payload_json, status, attempt_count, last_attempt_at, failure_reason, created_at) VALUES (?, 'report_ready', ?, ?, 'ready', 1, ?, NULL, ?)",
+                (order_id, channel, payload_json, utc_now_iso(), utc_now_iso()),
             )
             self._conn.commit()
         except sqlite3.IntegrityError:
             self._conn.rollback()
 
+    def mark_sent(self, order_id: str, event_type: str = "report_ready") -> None:
+        self._conn.execute(
+            "UPDATE delivery_notifications SET status='sent', last_attempt_at=?, failure_reason=NULL WHERE order_id=? AND event_type=?",
+            (utc_now_iso(), order_id, event_type),
+        )
+        self._conn.commit()
+
+    def mark_failed(
+        self,
+        order_id: str,
+        failure_reason: str,
+        *,
+        event_type: str = "report_ready",
+    ) -> None:
+        self._conn.execute(
+            "UPDATE delivery_notifications SET status='failed', attempt_count=attempt_count+1, last_attempt_at=?, failure_reason=? WHERE order_id=? AND event_type=?",
+            (utc_now_iso(), failure_reason, order_id, event_type),
+        )
+        self._conn.commit()
+
     def list_events(self, order_id: str) -> list[DeliveryNotificationEvent]:
         rows = self._conn.execute(
-            "SELECT order_id, event_type, channel, payload_json, created_at FROM delivery_notifications WHERE order_id=? ORDER BY id ASC",
+            "SELECT order_id, event_type, channel, payload_json, status, attempt_count, last_attempt_at, failure_reason, created_at FROM delivery_notifications WHERE order_id=? ORDER BY id ASC",
             (order_id,),
         ).fetchall()
         return [DeliveryNotificationEvent(**dict(row)) for row in rows]

docs/ACTIVE_EXECUTION_BOARD_2026-06-13.md

@@ -270,7 +270,7 @@ Owner: docs / PM
 
 Owner: data / engineer
 优先级: P1
-状态: in_progress
+状态: completed
 
 目标:
 

tests/test_delivery_notification.py

@@ -77,6 +77,9 @@ def test_report_ready_transition_creates_notification_event(
         notification_service.close()
     assert len(events) == 1
     assert events[0].event_type == "report_ready"
+    assert events[0].status == "ready"
+    assert events[0].attempt_count == 1
+    assert events[0].failure_reason is None
 
 
 def test_dao_delivered_transition_also_creates_notification_event(settings, tmp_path):
@@ -110,3 +113,49 @@ def test_dao_delivered_transition_also_creates_notification_event(settings, tmp_
         notification_service.close()
     assert len(events) == 1
     assert events[0].event_type == "report_ready"
+    assert events[0].status == "ready"
+    assert events[0].attempt_count == 1
+    assert events[0].failure_reason is None
+
+
+def test_delivery_notification_tracks_failure_and_sent_status(settings, tmp_path):
+    order = _seed_order(settings.orders_db_path, order_id="GKO-20260614-NOTIFY-STATUS")
+    _mark_paid(settings, order)
+    IntakeStore.for_db(settings.orders_db_path).save(
+        order_id=order.id, payload={"candidate_score": 578}, submit=True
+    )
+
+    report_path = tmp_path / "status-report.html"
+    pdf_path = tmp_path / "status-report.pdf"
+    report_path.write_text("<h1>status</h1>", encoding="utf-8")
+    pdf_path.write_bytes(b"%PDF-1.4\nstatus\n")
+
+    with OrdersDAO.connect(settings.orders_db_path) as dao:
+        dao.update(
+            order.id,
+            {"audit_report": str(report_path), "pdf_path": str(pdf_path)},
+            actor="test",
+            reason="attach_report",
+        )
+        dao.transition_status(order.id, "serving", actor="test", reason="processing")
+        dao.transition_status(
+            order.id, "delivered", actor="test", reason="report_ready"
+        )
+
+    notification_service = DeliveryNotificationService.for_db(settings.orders_db_path)
+    try:
+        notification_service.mark_failed(order.id, "smtp timeout")
+        failed = notification_service.list_events(order.id)[0]
+        assert failed.status == "failed"
+        assert failed.attempt_count == 2
+        assert failed.failure_reason == "smtp timeout"
+        assert failed.last_attempt_at is not None
+
+        notification_service.mark_sent(order.id)
+        sent = notification_service.list_events(order.id)[0]
+        assert sent.status == "sent"
+        assert sent.attempt_count == 2
+        assert sent.failure_reason is None
+        assert sent.last_attempt_at is not None
+    finally:
+        notification_service.close()