diff --git a/README.md b/README.md index 98886d0..81e32ff 100644 --- a/README.md +++ b/README.md @@ -55,6 +55,11 @@ cargo run -p shanghai-housing-api ```text GET /health GET /ready +GET /api/v1/data-sources +POST /api/v1/data-sources +GET /api/v1/ingestion-runs +POST /api/v1/ingestion-runs +PATCH /api/v1/ingestion-runs/{run_id}/finish GET /api/v1/areas/scores?month=2026-05 GET /api/v1/market/overview?month=2026-05 GET /api/v1/neighborhoods diff --git a/apps/api/src/handlers.rs b/apps/api/src/handlers.rs index b0bea3a..9eee70f 100644 --- a/apps/api/src/handlers.rs +++ b/apps/api/src/handlers.rs @@ -4,9 +4,10 @@ use serde::Serialize; use crate::error::{ApiError, ApiResult}; use crate::models::{ - AreaScore, AreaScoreSummary, CreateWatchlistItem, MarketOverview, MonthQuery, Neighborhood, - NeighborhoodQuery, NeighborhoodScore, NeighborhoodScoreQuery, UpdateWatchlistItem, - WatchlistItem, + AreaScore, AreaScoreSummary, CreateDataSource, CreateIngestionRun, CreateWatchlistItem, + DataSource, FinishIngestionRun, IngestionRun, IngestionRunQuery, MarketOverview, MonthQuery, + Neighborhood, NeighborhoodQuery, NeighborhoodScore, NeighborhoodScoreQuery, + UpdateWatchlistItem, WatchlistItem, }; use crate::state::AppState; @@ -176,6 +177,195 @@ pub async fn list_neighborhood_scores( Ok(Json(scores)) } +pub async fn list_data_sources(State(state): State) -> ApiResult>> { + let sources = sqlx::query_as::<_, DataSource>( + r#" + SELECT + source_id, + name, + source_type, + url, + cadence, + reliability, + notes, + created_at::text AS created_at + FROM audit.data_sources + ORDER BY name ASC + "#, + ) + .fetch_all(&state.pool) + .await?; + + Ok(Json(sources)) +} + +pub async fn create_data_source( + State(state): State, + Json(payload): Json, +) -> ApiResult> { + payload + .validate() + .map_err(|message| ApiError::BadRequest(message.to_string()))?; + + let source = sqlx::query_as::<_, DataSource>( + r#" + INSERT INTO audit.data_sources ( + name, + source_type, + url, + cadence, + reliability, + notes + ) + VALUES ($1, $2, $3, $4, COALESCE($5, 'unknown'), COALESCE($6, '')) + ON CONFLICT (name) DO UPDATE + SET source_type = EXCLUDED.source_type, + url = EXCLUDED.url, + cadence = EXCLUDED.cadence, + reliability = EXCLUDED.reliability, + notes = EXCLUDED.notes + RETURNING + source_id, + name, + source_type, + url, + cadence, + reliability, + notes, + created_at::text AS created_at + "#, + ) + .bind(payload.name.trim().to_string()) + .bind(payload.source_type.trim().to_string()) + .bind(payload.url) + .bind(payload.cadence) + .bind(payload.reliability) + .bind(payload.notes) + .fetch_one(&state.pool) + .await?; + + Ok(Json(source)) +} + +pub async fn list_ingestion_runs( + State(state): State, + Query(query): Query, +) -> ApiResult>> { + let runs = sqlx::query_as::<_, IngestionRun>( + r#" + SELECT + r.run_id, + r.source_id, + s.name AS source_name, + r.status, + r.started_at::text AS started_at, + r.finished_at::text AS finished_at, + r.raw_uri, + r.row_count, + r.error_message + FROM audit.ingestion_runs r + LEFT JOIN audit.data_sources s ON s.source_id = r.source_id + WHERE ($1::bigint IS NULL OR r.source_id = $1) + ORDER BY r.started_at DESC, r.run_id DESC + LIMIT 100 + "#, + ) + .bind(query.source_id) + .fetch_all(&state.pool) + .await?; + + Ok(Json(runs)) +} + +pub async fn create_ingestion_run( + State(state): State, + Json(payload): Json, +) -> ApiResult> { + payload + .validate() + .map_err(|message| ApiError::BadRequest(message.to_string()))?; + + let run = sqlx::query_as::<_, IngestionRun>( + r#" + INSERT INTO audit.ingestion_runs ( + source_id, + status, + raw_uri, + row_count, + error_message + ) + VALUES ($1, COALESCE($2, 'running'), $3, $4, $5) + RETURNING + run_id, + source_id, + ( + SELECT name + FROM audit.data_sources + WHERE source_id = audit.ingestion_runs.source_id + ) AS source_name, + status, + started_at::text AS started_at, + finished_at::text AS finished_at, + raw_uri, + row_count, + error_message + "#, + ) + .bind(payload.source_id) + .bind(payload.status) + .bind(payload.raw_uri) + .bind(payload.row_count) + .bind(payload.error_message) + .fetch_one(&state.pool) + .await?; + + Ok(Json(run)) +} + +pub async fn finish_ingestion_run( + State(state): State, + Path(run_id): Path, + Json(payload): Json, +) -> ApiResult> { + payload + .validate() + .map_err(|message| ApiError::BadRequest(message.to_string()))?; + + let run = sqlx::query_as::<_, IngestionRun>( + r#" + UPDATE audit.ingestion_runs + SET status = $2, + finished_at = now(), + row_count = COALESCE($3, row_count), + error_message = $4 + WHERE run_id = $1 + RETURNING + run_id, + source_id, + ( + SELECT name + FROM audit.data_sources + WHERE source_id = audit.ingestion_runs.source_id + ) AS source_name, + status, + started_at::text AS started_at, + finished_at::text AS finished_at, + raw_uri, + row_count, + error_message + "#, + ) + .bind(run_id) + .bind(payload.status) + .bind(payload.row_count) + .bind(payload.error_message) + .fetch_optional(&state.pool) + .await? + .ok_or_else(|| ApiError::BadRequest("ingestion run not found".to_string()))?; + + Ok(Json(run)) +} + pub async fn list_watchlist_items( State(state): State, ) -> ApiResult>> { @@ -367,7 +557,10 @@ fn validate_month(month: &str) -> ApiResult<()> { #[cfg(test)] mod tests { - use crate::models::{CreateWatchlistItem, UpdateWatchlistItem}; + use crate::models::{ + CreateDataSource, CreateIngestionRun, CreateWatchlistItem, FinishIngestionRun, + UpdateWatchlistItem, + }; use super::validate_month; @@ -430,4 +623,79 @@ mod tests { .validate() .is_ok()); } + + #[test] + fn validates_data_source_payload() { + assert!(CreateDataSource { + name: "".to_string(), + source_type: "official".to_string(), + url: None, + cadence: None, + reliability: None, + notes: None, + } + .validate() + .is_err()); + + assert!(CreateDataSource { + name: "上海市统计局".to_string(), + source_type: "official".to_string(), + url: Some("https://tjj.sh.gov.cn/".to_string()), + cadence: Some("monthly".to_string()), + reliability: Some("high".to_string()), + notes: None, + } + .validate() + .is_ok()); + + assert!(CreateDataSource { + name: "unsafe".to_string(), + source_type: "manual".to_string(), + url: None, + cadence: None, + reliability: None, + notes: Some("contains password marker".to_string()), + } + .validate() + .is_err()); + } + + #[test] + fn validates_ingestion_run_payloads() { + assert!(CreateIngestionRun { + source_id: Some(1), + status: Some("running".to_string()), + raw_uri: Some("raw/official/2026-05.csv".to_string()), + row_count: Some(10), + error_message: None, + } + .validate() + .is_ok()); + + assert!(CreateIngestionRun { + source_id: Some(1), + status: Some("unknown".to_string()), + raw_uri: None, + row_count: None, + error_message: None, + } + .validate() + .is_err()); + + assert!(FinishIngestionRun { + status: "succeeded".to_string(), + row_count: Some(10), + error_message: None, + } + .validate() + .is_ok()); + + assert!(FinishIngestionRun { + status: "running".to_string(), + row_count: None, + error_message: None, + } + .validate() + .is_err()); + } } diff --git a/apps/api/src/models.rs b/apps/api/src/models.rs index 31ea62a..544ae29 100644 --- a/apps/api/src/models.rs +++ b/apps/api/src/models.rs @@ -17,6 +17,11 @@ pub struct NeighborhoodScoreQuery { pub area_id: Option, } +#[derive(Debug, Deserialize)] +pub struct IngestionRunQuery { + pub source_id: Option, +} + #[derive(Debug, Clone, Serialize, FromRow)] pub struct AreaScore { pub area_id: String, @@ -118,6 +123,123 @@ pub struct NeighborhoodScore { pub available_units: i32, } +#[derive(Debug, Clone, Serialize, FromRow)] +pub struct DataSource { + pub source_id: i64, + pub name: String, + pub source_type: String, + pub url: Option, + pub cadence: Option, + pub reliability: String, + pub notes: String, + pub created_at: String, +} + +#[derive(Debug, Deserialize)] +pub struct CreateDataSource { + pub name: String, + pub source_type: String, + pub url: Option, + pub cadence: Option, + pub reliability: Option, + pub notes: Option, +} + +impl CreateDataSource { + pub fn validate(&self) -> Result<(), &'static str> { + if self.name.trim().is_empty() { + return Err("name is required"); + } + if self.source_type.trim().is_empty() { + return Err("source_type is required"); + } + for value in [ + Some(self.name.as_str()), + Some(self.source_type.as_str()), + self.url.as_deref(), + self.cadence.as_deref(), + self.reliability.as_deref(), + self.notes.as_deref(), + ] + .into_iter() + .flatten() + { + if has_sensitive_text(value) { + return Err("payload contains sensitive text"); + } + } + Ok(()) + } +} + +#[derive(Debug, Clone, Serialize, FromRow)] +pub struct IngestionRun { + pub run_id: i64, + pub source_id: Option, + pub source_name: Option, + pub status: String, + pub started_at: String, + pub finished_at: Option, + pub raw_uri: Option, + pub row_count: Option, + pub error_message: Option, +} + +#[derive(Debug, Deserialize)] +pub struct CreateIngestionRun { + pub source_id: Option, + pub status: Option, + pub raw_uri: Option, + pub row_count: Option, + pub error_message: Option, +} + +impl CreateIngestionRun { + pub fn validate(&self) -> Result<(), &'static str> { + if let Some(status) = &self.status { + if !is_ingestion_status(status) { + return Err("status must be running, succeeded, or failed"); + } + } + if matches!(self.row_count, Some(value) if value < 0) { + return Err("row_count must be non-negative"); + } + for value in [self.raw_uri.as_deref(), self.error_message.as_deref()] + .into_iter() + .flatten() + { + if has_sensitive_text(value) { + return Err("payload contains sensitive text"); + } + } + Ok(()) + } +} + +#[derive(Debug, Deserialize)] +pub struct FinishIngestionRun { + pub status: String, + pub row_count: Option, + pub error_message: Option, +} + +impl FinishIngestionRun { + pub fn validate(&self) -> Result<(), &'static str> { + if !matches!(self.status.as_str(), "succeeded" | "failed") { + return Err("status must be succeeded or failed"); + } + if matches!(self.row_count, Some(value) if value < 0) { + return Err("row_count must be non-negative"); + } + if let Some(value) = self.error_message.as_deref() { + if has_sensitive_text(value) { + return Err("payload contains sensitive text"); + } + } + Ok(()) + } +} + #[derive(Debug, Clone, Serialize, FromRow)] pub struct WatchlistItem { pub watchlist_item_id: i64, @@ -148,6 +270,21 @@ impl CreateWatchlistItem { } } +fn is_ingestion_status(status: &str) -> bool { + matches!(status, "running" | "succeeded" | "failed") +} + +fn has_sensitive_text(value: &str) -> bool { + let lower = value.to_ascii_lowercase(); + lower.contains("password") + || lower.contains("secret") + || lower.contains("postgres://") + || lower.contains("root_") + || lower.contains("database_url") + || value.starts_with("/Users/") + || value.starts_with("/private/") +} + #[derive(Debug, Deserialize)] pub struct UpdateWatchlistItem { pub target_price_psm: Option, diff --git a/apps/api/src/routes.rs b/apps/api/src/routes.rs index a249fa2..1b1338c 100644 --- a/apps/api/src/routes.rs +++ b/apps/api/src/routes.rs @@ -4,9 +4,10 @@ use tower_http::cors::CorsLayer; use tower_http::trace::TraceLayer; use crate::handlers::{ - archive_watchlist_item, create_watchlist_item, get_neighborhood, health, list_area_scores, - list_neighborhood_scores, list_neighborhoods, list_watchlist_items, market_overview, ready, - update_watchlist_item, + archive_watchlist_item, create_data_source, create_ingestion_run, create_watchlist_item, + finish_ingestion_run, get_neighborhood, health, list_area_scores, list_data_sources, + list_ingestion_runs, list_neighborhood_scores, list_neighborhoods, list_watchlist_items, + market_overview, ready, update_watchlist_item, }; use crate::state::AppState; @@ -18,6 +19,18 @@ pub fn build_router(state: AppState) -> Router { "/api/v1", Router::new() .route("/areas/scores", get(list_area_scores)) + .route( + "/data-sources", + get(list_data_sources).post(create_data_source), + ) + .route( + "/ingestion-runs", + get(list_ingestion_runs).post(create_ingestion_run), + ) + .route( + "/ingestion-runs/{run_id}/finish", + patch(finish_ingestion_run), + ) .route("/market/overview", get(market_overview)) .route("/neighborhoods", get(list_neighborhoods)) .route("/neighborhoods/scores", get(list_neighborhood_scores)) diff --git a/docs/product_roadmap.md b/docs/product_roadmap.md index 65a5e41..0b42339 100644 --- a/docs/product_roadmap.md +++ b/docs/product_roadmap.md @@ -32,6 +32,8 @@ ### M1.1 数据源登记与导入批次审计 +状态:已完成。 + 目标: - 建立数据源目录的后台 API。