From e5d95c05987569fe52263ebb17040d16628b1411 Mon Sep 17 00:00:00 2001 From: shay7sev Date: Wed, 24 Jun 2026 09:55:47 +0800 Subject: [PATCH] feat: add import execution api --- README.md | 1 + .../202606240002_import_execution.sql | 30 ++ apps/api/src/handlers.rs | 59 ++- apps/api/src/import_templates.rs | 403 +++++++++++++++++- apps/api/src/models.rs | 22 + apps/api/src/routes.rs | 9 +- apps/web/src/lib/api.ts | 29 ++ docs/import_templates/README.md | 38 ++ docs/product_roadmap.md | 6 +- 9 files changed, 584 insertions(+), 13 deletions(-) create mode 100644 apps/api/migrations/202606240002_import_execution.sql diff --git a/README.md b/README.md index 7e7eeb2..e50d780 100644 --- a/README.md +++ b/README.md @@ -63,6 +63,7 @@ PATCH /api/v1/ingestion-runs/{run_id}/finish GET /api/v1/raw-artifacts POST /api/v1/raw-artifacts POST /api/v1/imports/validate +POST /api/v1/imports/execute 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/migrations/202606240002_import_execution.sql b/apps/api/migrations/202606240002_import_execution.sql new file mode 100644 index 0000000..a3ee1d8 --- /dev/null +++ b/apps/api/migrations/202606240002_import_execution.sql @@ -0,0 +1,30 @@ +CREATE TABLE IF NOT EXISTS raw.import_rows ( + import_row_id BIGSERIAL PRIMARY KEY, + run_id BIGINT NOT NULL REFERENCES audit.ingestion_runs(run_id), + raw_artifact_id BIGINT NOT NULL REFERENCES audit.raw_artifacts(artifact_id), + template_name TEXT NOT NULL, + target_table TEXT NOT NULL, + row_number INTEGER NOT NULL CHECK (row_number >= 2), + payload JSONB NOT NULL, + imported_at TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE (run_id, template_name, row_number) +); + +ALTER TABLE silver.area_monthly_metrics + ADD COLUMN IF NOT EXISTS ingestion_run_id BIGINT REFERENCES audit.ingestion_runs(run_id), + ADD COLUMN IF NOT EXISTS raw_artifact_id BIGINT REFERENCES audit.raw_artifacts(artifact_id); + +ALTER TABLE silver.neighborhood_monthly_metrics + ADD COLUMN IF NOT EXISTS ingestion_run_id BIGINT REFERENCES audit.ingestion_runs(run_id), + ADD COLUMN IF NOT EXISTS raw_artifact_id BIGINT REFERENCES audit.raw_artifacts(artifact_id); + +CREATE INDEX IF NOT EXISTS idx_import_rows_run_id + ON raw.import_rows(run_id); +CREATE INDEX IF NOT EXISTS idx_import_rows_raw_artifact_id + ON raw.import_rows(raw_artifact_id); +CREATE INDEX IF NOT EXISTS idx_import_rows_template_name + ON raw.import_rows(template_name, imported_at DESC); +CREATE INDEX IF NOT EXISTS idx_area_metrics_ingestion_run + ON silver.area_monthly_metrics(ingestion_run_id); +CREATE INDEX IF NOT EXISTS idx_neighborhood_metrics_ingestion_run + ON silver.neighborhood_monthly_metrics(ingestion_run_id); diff --git a/apps/api/src/handlers.rs b/apps/api/src/handlers.rs index 3faa634..9c8d9ed 100644 --- a/apps/api/src/handlers.rs +++ b/apps/api/src/handlers.rs @@ -3,13 +3,14 @@ use axum::Json; use serde::Serialize; use crate::error::{ApiError, ApiResult}; -use crate::import_templates::validate_import_payload; +use crate::import_templates::{execute_import_payload, validate_import_payload}; use crate::models::{ AreaScore, AreaScoreSummary, CreateDataSource, CreateIngestionRun, CreateRawArtifact, - CreateWatchlistItem, DataSource, FinishIngestionRun, ImportValidationRequest, - ImportValidationResponse, IngestionRun, IngestionRunQuery, MarketOverview, MonthQuery, - Neighborhood, NeighborhoodQuery, NeighborhoodScore, NeighborhoodScoreQuery, RawArtifact, - RawArtifactQuery, UpdateWatchlistItem, WatchlistItem, + CreateWatchlistItem, DataSource, FinishIngestionRun, ImportExecutionRequest, + ImportExecutionResponse, ImportValidationRequest, ImportValidationResponse, IngestionRun, + IngestionRunQuery, MarketOverview, MonthQuery, Neighborhood, NeighborhoodQuery, + NeighborhoodScore, NeighborhoodScoreQuery, RawArtifact, RawArtifactQuery, UpdateWatchlistItem, + WatchlistItem, }; use crate::state::AppState; @@ -41,6 +42,54 @@ pub async fn validate_import( Ok(Json(result)) } +pub async fn execute_import( + State(state): State, + Json(payload): Json, +) -> ApiResult> { + let mut tx = state.pool.begin().await?; + let result = execute_import_payload(&mut tx, &payload).await; + + match result { + Ok(response) => { + sqlx::query( + r#" + UPDATE audit.ingestion_runs + SET status = 'succeeded', + finished_at = now(), + row_count = $2, + raw_artifact_id = $3, + error_message = NULL + WHERE run_id = $1 + "#, + ) + .bind(payload.run_id) + .bind(response.row_count as i32) + .bind(payload.raw_artifact_id) + .execute(&mut *tx) + .await?; + tx.commit().await?; + Ok(Json(response)) + } + Err(message) => { + tx.rollback().await?; + sqlx::query( + r#" + UPDATE audit.ingestion_runs + SET status = 'failed', + finished_at = now(), + error_message = $2 + WHERE run_id = $1 + "#, + ) + .bind(payload.run_id) + .bind(message.clone()) + .execute(&state.pool) + .await?; + Err(ApiError::BadRequest(message)) + } + } +} + pub async fn list_area_scores( State(state): State, Query(query): Query, diff --git a/apps/api/src/import_templates.rs b/apps/api/src/import_templates.rs index 81dcc73..b9c748c 100644 --- a/apps/api/src/import_templates.rs +++ b/apps/api/src/import_templates.rs @@ -3,12 +3,18 @@ use std::collections::{HashMap, HashSet}; use serde::Deserialize; use serde_json::Value; -use crate::models::{ImportValidationError, ImportValidationRequest, ImportValidationResponse}; +use sqlx::{Postgres, Transaction}; + +use crate::models::{ + ImportExecutionRequest, ImportExecutionResponse, ImportValidationError, + ImportValidationRequest, ImportValidationResponse, +}; const IMPORT_TEMPLATE_SPEC: &str = include_str!("../../../config/import_templates.json"); #[derive(Debug, Deserialize)] struct TemplateSpec { + target_table: String, columns: Vec, } @@ -100,11 +106,361 @@ pub fn validate_import_payload( }) } +pub async fn execute_import_payload( + tx: &mut Transaction<'_, Postgres>, + payload: &ImportExecutionRequest, +) -> Result { + let validation_request = ImportValidationRequest { + template_name: payload.template_name.clone(), + rows: payload.rows.clone(), + }; + let validation = validate_import_payload(&validation_request)?; + if !validation.valid { + let summary = validation + .errors + .iter() + .take(5) + .map(|error| format!("row {}, {}: {}", error.row, error.column, error.message)) + .collect::>() + .join("; "); + return Err(format!("import validation failed: {summary}")); + } + + let spec = load_specs()? + .remove(&payload.template_name) + .ok_or_else(|| format!("unknown template '{}'", payload.template_name))?; + ensure_supported_execution_template(&payload.template_name)?; + ensure_run_artifact_link(tx, payload.run_id, payload.raw_artifact_id).await?; + + let previous_run_ids = + find_previous_import_runs(tx, payload.raw_artifact_id, payload.run_id).await?; + let duplicate_artifact = !previous_run_ids.is_empty(); + if duplicate_artifact && !payload.allow_reimport.unwrap_or(false) { + return Err(format!( + "raw artifact has already been imported by run(s): {}", + previous_run_ids + .iter() + .map(i64::to_string) + .collect::>() + .join(", ") + )); + } + + let raw_row_count = insert_raw_rows(tx, payload, &spec.target_table).await?; + upsert_silver_rows(tx, payload).await?; + + Ok(ImportExecutionResponse { + template_name: payload.template_name.clone(), + target_table: spec.target_table, + run_id: payload.run_id, + raw_artifact_id: payload.raw_artifact_id, + row_count: payload.rows.len(), + raw_row_count, + duplicate_artifact, + previous_run_ids, + status: "succeeded".to_string(), + }) +} + fn load_specs() -> Result, String> { serde_json::from_str(IMPORT_TEMPLATE_SPEC) .map_err(|error| format!("failed to parse import template specs: {error}")) } +async fn ensure_run_artifact_link( + tx: &mut Transaction<'_, Postgres>, + run_id: i64, + raw_artifact_id: i64, +) -> Result<(), String> { + let linked_artifact_id = sqlx::query_scalar::<_, Option>( + r#" + SELECT raw_artifact_id + FROM audit.ingestion_runs + WHERE run_id = $1 + "#, + ) + .bind(run_id) + .fetch_optional(&mut **tx) + .await + .map_err(|error| format!("failed to fetch ingestion run: {error}"))? + .ok_or_else(|| "ingestion run not found".to_string())?; + + let artifact_exists = sqlx::query_scalar::<_, bool>( + r#" + SELECT EXISTS ( + SELECT 1 + FROM audit.raw_artifacts + WHERE artifact_id = $1 + ) + "#, + ) + .bind(raw_artifact_id) + .fetch_one(&mut **tx) + .await + .map_err(|error| format!("failed to fetch raw artifact: {error}"))?; + if !artifact_exists { + return Err("raw artifact not found".to_string()); + } + + if let Some(existing_artifact_id) = linked_artifact_id { + if existing_artifact_id != raw_artifact_id { + return Err("ingestion run is linked to a different raw artifact".to_string()); + } + } else { + sqlx::query( + r#" + UPDATE audit.ingestion_runs + SET raw_artifact_id = $2, + raw_uri = COALESCE( + raw_uri, + ( + SELECT raw_uri + FROM audit.raw_artifacts + WHERE artifact_id = $2 + ) + ) + WHERE run_id = $1 + "#, + ) + .bind(run_id) + .bind(raw_artifact_id) + .execute(&mut **tx) + .await + .map_err(|error| format!("failed to link ingestion run to raw artifact: {error}"))?; + } + + Ok(()) +} + +async fn find_previous_import_runs( + tx: &mut Transaction<'_, Postgres>, + raw_artifact_id: i64, + current_run_id: i64, +) -> Result, String> { + sqlx::query_scalar::<_, i64>( + r#" + SELECT DISTINCT run_id + FROM raw.import_rows + WHERE raw_artifact_id = $1 + AND run_id <> $2 + ORDER BY run_id + "#, + ) + .bind(raw_artifact_id) + .bind(current_run_id) + .fetch_all(&mut **tx) + .await + .map_err(|error| format!("failed to check duplicate raw artifact imports: {error}")) +} + +async fn insert_raw_rows( + tx: &mut Transaction<'_, Postgres>, + payload: &ImportExecutionRequest, + target_table: &str, +) -> Result { + let mut inserted = 0; + for (index, row) in payload.rows.iter().enumerate() { + let result = sqlx::query( + r#" + INSERT INTO raw.import_rows ( + run_id, + raw_artifact_id, + template_name, + target_table, + row_number, + payload + ) + VALUES ($1, $2, $3, $4, $5, $6) + ON CONFLICT (run_id, template_name, row_number) DO UPDATE + SET payload = EXCLUDED.payload, + raw_artifact_id = EXCLUDED.raw_artifact_id, + target_table = EXCLUDED.target_table, + imported_at = now() + "#, + ) + .bind(payload.run_id) + .bind(payload.raw_artifact_id) + .bind(&payload.template_name) + .bind(target_table) + .bind((index + 2) as i32) + .bind(Value::Object(row.clone().into_iter().collect())) + .execute(&mut **tx) + .await + .map_err(|error| format!("failed to insert raw import row {}: {error}", index + 2))?; + inserted += result.rows_affected(); + } + Ok(inserted) +} + +async fn upsert_silver_rows( + tx: &mut Transaction<'_, Postgres>, + payload: &ImportExecutionRequest, +) -> Result<(), String> { + match payload.template_name.as_str() { + "area_monthly_metrics" => upsert_area_monthly_metrics(tx, payload).await, + "neighborhood_monthly_metrics" => upsert_neighborhood_monthly_metrics(tx, payload).await, + _ => Err(format!( + "template '{}' is not supported by import execution", + payload.template_name + )), + } +} + +async fn upsert_area_monthly_metrics( + tx: &mut Transaction<'_, Postgres>, + payload: &ImportExecutionRequest, +) -> Result<(), String> { + for (index, row) in payload.rows.iter().enumerate() { + sqlx::query( + r#" + INSERT INTO silver.area_monthly_metrics ( + area_id, + month, + transaction_count, + transaction_price_psm, + listing_count, + listing_price_psm, + median_days_on_market, + rent_price_psm, + new_supply_units, + land_residential_gfa_sqm, + mortgage_rate_pct, + policy_signal, + source, + ingestion_run_id, + raw_artifact_id + ) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15) + ON CONFLICT (area_id, month) DO UPDATE + SET transaction_count = EXCLUDED.transaction_count, + transaction_price_psm = EXCLUDED.transaction_price_psm, + listing_count = EXCLUDED.listing_count, + listing_price_psm = EXCLUDED.listing_price_psm, + median_days_on_market = EXCLUDED.median_days_on_market, + rent_price_psm = EXCLUDED.rent_price_psm, + new_supply_units = EXCLUDED.new_supply_units, + land_residential_gfa_sqm = EXCLUDED.land_residential_gfa_sqm, + mortgage_rate_pct = EXCLUDED.mortgage_rate_pct, + policy_signal = EXCLUDED.policy_signal, + source = EXCLUDED.source, + ingestion_run_id = EXCLUDED.ingestion_run_id, + raw_artifact_id = EXCLUDED.raw_artifact_id, + updated_at = now() + "#, + ) + .bind(required_text(row, "area_id", index)?) + .bind(required_text(row, "month", index)?) + .bind(required_i32(row, "transaction_count", index)?) + .bind(required_f64(row, "transaction_price_psm", index)?) + .bind(required_i32(row, "listing_count", index)?) + .bind(required_f64(row, "listing_price_psm", index)?) + .bind(required_f64(row, "median_days_on_market", index)?) + .bind(required_f64(row, "rent_price_psm", index)?) + .bind(required_i32(row, "new_supply_units", index)?) + .bind(required_f64(row, "land_residential_gfa_sqm", index)?) + .bind(required_f64(row, "mortgage_rate_pct", index)?) + .bind(required_i32(row, "policy_signal", index)?) + .bind(required_text(row, "source", index)?) + .bind(payload.run_id) + .bind(payload.raw_artifact_id) + .execute(&mut **tx) + .await + .map_err(|error| format!("failed to upsert area row {}: {error}", index + 2))?; + } + Ok(()) +} + +async fn upsert_neighborhood_monthly_metrics( + tx: &mut Transaction<'_, Postgres>, + payload: &ImportExecutionRequest, +) -> Result<(), String> { + for (index, row) in payload.rows.iter().enumerate() { + sqlx::query( + r#" + INSERT INTO silver.neighborhood_monthly_metrics ( + neighborhood_id, + month, + transaction_count, + transaction_price_psm, + listing_count, + listing_price_psm, + rent_price_psm, + median_days_on_market, + available_units, + source, + ingestion_run_id, + raw_artifact_id + ) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) + ON CONFLICT (neighborhood_id, month) DO UPDATE + SET transaction_count = EXCLUDED.transaction_count, + transaction_price_psm = EXCLUDED.transaction_price_psm, + listing_count = EXCLUDED.listing_count, + listing_price_psm = EXCLUDED.listing_price_psm, + rent_price_psm = EXCLUDED.rent_price_psm, + median_days_on_market = EXCLUDED.median_days_on_market, + available_units = EXCLUDED.available_units, + source = EXCLUDED.source, + ingestion_run_id = EXCLUDED.ingestion_run_id, + raw_artifact_id = EXCLUDED.raw_artifact_id, + updated_at = now() + "#, + ) + .bind(required_text(row, "neighborhood_id", index)?) + .bind(required_text(row, "month", index)?) + .bind(required_i32(row, "transaction_count", index)?) + .bind(required_f64(row, "transaction_price_psm", index)?) + .bind(required_i32(row, "listing_count", index)?) + .bind(required_f64(row, "listing_price_psm", index)?) + .bind(required_f64(row, "rent_price_psm", index)?) + .bind(required_f64(row, "median_days_on_market", index)?) + .bind(required_i32(row, "available_units", index)?) + .bind(required_text(row, "source", index)?) + .bind(payload.run_id) + .bind(payload.raw_artifact_id) + .execute(&mut **tx) + .await + .map_err(|error| format!("failed to upsert neighborhood row {}: {error}", index + 2))?; + } + Ok(()) +} + +fn ensure_supported_execution_template(template_name: &str) -> Result<(), String> { + match template_name { + "area_monthly_metrics" | "neighborhood_monthly_metrics" => Ok(()), + _ => Err(format!( + "template '{template_name}' is not supported by import execution" + )), + } +} + +fn required_text( + row: &HashMap, + column: &str, + index: usize, +) -> Result { + let value = normalize_value(row.get(column)); + if value.is_empty() { + Err(format!("row {}, column {column} is required", index + 2)) + } else { + Ok(value) + } +} + +fn required_i32(row: &HashMap, column: &str, index: usize) -> Result { + let value = required_text(row, column, index)?; + value + .parse::() + .map_err(|_| format!("row {}, column {column} must be an integer", index + 2)) +} + +fn required_f64(row: &HashMap, column: &str, index: usize) -> Result { + let value = required_text(row, column, index)?; + value + .parse::() + .map_err(|_| format!("row {}, column {column} must be a number", index + 2)) +} + fn normalize_value(value: Option<&Value>) -> String { match value { None | Some(Value::Null) => String::new(), @@ -311,9 +667,9 @@ mod tests { use serde_json::json; - use crate::models::ImportValidationRequest; + use crate::models::{ImportExecutionRequest, ImportValidationRequest}; - use super::{is_valid_date, validate_import_payload}; + use super::{ensure_supported_execution_template, is_valid_date, validate_import_payload}; #[test] fn validates_good_import_rows() { @@ -386,4 +742,45 @@ mod tests { assert!(is_valid_date("2024-02-29")); assert!(!is_valid_date("2026-13-01")); } + + #[test] + fn limits_execution_templates_to_metric_imports() { + assert!(ensure_supported_execution_template("area_monthly_metrics").is_ok()); + assert!(ensure_supported_execution_template("neighborhood_monthly_metrics").is_ok()); + assert!(ensure_supported_execution_template("policy_events").is_err()); + } + + #[test] + fn execution_payload_rows_reuse_validation_rules() { + let request = ImportExecutionRequest { + template_name: "area_monthly_metrics".to_string(), + run_id: 1, + raw_artifact_id: 1, + allow_reimport: None, + rows: vec![HashMap::from([ + ("area_id".to_string(), json!("qiantan")), + ("month".to_string(), json!("2026-05")), + ("transaction_count".to_string(), json!("93")), + ("transaction_price_psm".to_string(), json!("122500")), + ("listing_count".to_string(), json!("380")), + ("listing_price_psm".to_string(), json!("126500")), + ("median_days_on_market".to_string(), json!("53")), + ("rent_price_psm".to_string(), json!("192")), + ("new_supply_units".to_string(), json!("60")), + ("land_residential_gfa_sqm".to_string(), json!("0")), + ("mortgage_rate_pct".to_string(), json!("3.35")), + ("policy_signal".to_string(), json!("1")), + ("source".to_string(), json!("manual_research")), + ])], + }; + + let validation = validate_import_payload(&ImportValidationRequest { + template_name: request.template_name, + rows: request.rows, + }) + .expect("valid import spec"); + + assert!(validation.valid); + assert_eq!(validation.row_count, 1); + } } diff --git a/apps/api/src/models.rs b/apps/api/src/models.rs index 51c35ab..7d8cc06 100644 --- a/apps/api/src/models.rs +++ b/apps/api/src/models.rs @@ -63,6 +63,28 @@ pub struct ImportValidationResponse { pub errors: Vec, } +#[derive(Debug, Deserialize)] +pub struct ImportExecutionRequest { + pub template_name: String, + pub run_id: i64, + pub raw_artifact_id: i64, + pub rows: Vec>, + pub allow_reimport: Option, +} + +#[derive(Debug, Serialize)] +pub struct ImportExecutionResponse { + pub template_name: String, + pub target_table: String, + pub run_id: i64, + pub raw_artifact_id: i64, + pub row_count: usize, + pub raw_row_count: u64, + pub duplicate_artifact: bool, + pub previous_run_ids: Vec, + pub status: String, +} + #[derive(Debug, Clone, Serialize, FromRow)] pub struct AreaScore { pub area_id: String, diff --git a/apps/api/src/routes.rs b/apps/api/src/routes.rs index c0d3c1d..97ff251 100644 --- a/apps/api/src/routes.rs +++ b/apps/api/src/routes.rs @@ -5,10 +5,10 @@ use tower_http::trace::TraceLayer; use crate::handlers::{ archive_watchlist_item, create_data_source, create_ingestion_run, create_raw_artifact, - create_watchlist_item, finish_ingestion_run, get_neighborhood, health, list_area_scores, - list_data_sources, list_ingestion_runs, list_neighborhood_scores, list_neighborhoods, - list_raw_artifacts, list_watchlist_items, market_overview, ready, update_watchlist_item, - validate_import, + create_watchlist_item, execute_import, finish_ingestion_run, get_neighborhood, health, + list_area_scores, list_data_sources, list_ingestion_runs, list_neighborhood_scores, + list_neighborhoods, list_raw_artifacts, list_watchlist_items, market_overview, ready, + update_watchlist_item, validate_import, }; use crate::state::AppState; @@ -37,6 +37,7 @@ pub fn build_router(state: AppState) -> Router { get(list_raw_artifacts).post(create_raw_artifact), ) .route("/imports/validate", axum::routing::post(validate_import)) + .route("/imports/execute", axum::routing::post(execute_import)) .route("/market/overview", get(market_overview)) .route("/neighborhoods", get(list_neighborhoods)) .route("/neighborhoods/scores", get(list_neighborhood_scores)) diff --git a/apps/web/src/lib/api.ts b/apps/web/src/lib/api.ts index a45678b..3aebd1b 100644 --- a/apps/web/src/lib/api.ts +++ b/apps/web/src/lib/api.ts @@ -160,6 +160,26 @@ export type CreateRawArtifact = { notes?: string; }; +export type ImportExecutionRequest = { + template_name: string; + run_id: number; + raw_artifact_id: number; + rows: Record[]; + allow_reimport?: boolean; +}; + +export type ImportExecutionResponse = { + template_name: string; + target_table: string; + run_id: number; + raw_artifact_id: number; + row_count: number; + raw_row_count: number; + duplicate_artifact: boolean; + previous_run_ids: number[]; + status: string; +}; + const API_BASE_URL = process.env.NEXT_PUBLIC_API_BASE_URL?.replace(/\/$/, "") ?? "http://127.0.0.1:8080"; @@ -237,6 +257,15 @@ export async function createRawArtifact(payload: CreateRawArtifact): Promise { + return fetchJson("/api/v1/imports/execute", { + method: "POST", + body: JSON.stringify(payload), + }); +} + async function fetchJson(path: string, init?: RequestInit): Promise { const response = await fetch(`${API_BASE_URL}${path}`, { ...init, diff --git a/docs/import_templates/README.md b/docs/import_templates/README.md index b682bf1..779e3ec 100644 --- a/docs/import_templates/README.md +++ b/docs/import_templates/README.md @@ -102,3 +102,41 @@ curl -X POST http://127.0.0.1:8080/api/v1/raw-artifacts \ ``` 同一个 `sha256` 重复登记时，API 返回已有 artifact，并可将新的导入批次关联到该 artifact。 + +## API 执行导入 + +执行顺序： + +1. 创建或选择数据源。 +2. 创建 `audit.ingestion_runs` 批次。 +3. 使用文件指纹登记 `audit.raw_artifacts`。 +4. 调用执行接口写入 `raw.import_rows` 和对应 `silver` 表。 + +```bash +curl -X POST http://127.0.0.1:8080/api/v1/imports/execute \ + -H 'content-type: application/json' \ + -d '{ + "template_name": "area_monthly_metrics", + "run_id": 1, + "raw_artifact_id": 1, + "rows": [ + { + "area_id": "qiantan", + "month": "2026-05", + "transaction_count": "93", + "transaction_price_psm": "122500", + "listing_count": "380", + "listing_price_psm": "126500", + "median_days_on_market": "53", + "rent_price_psm": "192", + "new_supply_units": "60", + "land_residential_gfa_sqm": "0", + "mortgage_rate_pct": "3.35", + "policy_signal": "1", + "source": "manual_research" + } + ] + }' +``` + +当前执行接口支持 `area_monthly_metrics` 和 `neighborhood_monthly_metrics`。校验失败时不会写入业务表，并会把导入批次标记为 `failed`。 diff --git a/docs/product_roadmap.md b/docs/product_roadmap.md index b8a44bb..18b41e9 100644 --- a/docs/product_roadmap.md +++ b/docs/product_roadmap.md @@ -122,6 +122,8 @@ ### M1.5 PostgreSQL 导入执行接口 +状态：已完成。 + 目标： - 将已校验模板数据正式写入 PostgreSQL 的 `raw`/`silver`/`audit`。 @@ -129,7 +131,8 @@ 交付物： -- PostgreSQL 导入执行 API。 +- `POST /api/v1/imports/execute` +- `raw.import_rows` 原始行 JSON 留痕。 - 支持板块月度指标和小区月度指标的 upsert。 - 导入失败时写入明确错误并保留审计记录。 @@ -138,6 +141,7 @@ - 校验失败不写入业务表。 - 导入成功后可通过批次追溯到原始文件哈希。 - 重复导入同一文件时能识别并提示。 +- 当前执行范围：`area_monthly_metrics`、`neighborhood_monthly_metrics`。 ## M2 资产研究工作台