Tools/huheng-research
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add import execution api

Browse Source
This commit is contained in:
shay7sev
2026-06-24 09:55:47 +08:00
parent a770070b34
commit e5d95c0598
9 changed files with 584 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
README.md
View File

@@ -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

30
apps/api/migrations/202606240002_import_execution.sql Normal file
View File

@@ -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);

59
apps/api/src/handlers.rs
View File

@@ -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<AppState>,
Json(payload): Json<ImportExecutionRequest>,
) -> ApiResult<Json<ImportExecutionResponse>> {
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<AppState>,
Query(query): Query<MonthQuery>,

403
apps/api/src/import_templates.rs
View File

@@ -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<ColumnSpec>,
}
@@ -100,11 +106,361 @@ pub fn validate_import_payload(
})
}
pub async fn execute_import_payload(
tx: &mut Transaction<'_, Postgres>,
payload: &ImportExecutionRequest,
) -> Result<ImportExecutionResponse, String> {
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::<Vec<_>>()
.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::<Vec<_>>()
.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<HashMap<String, TemplateSpec>, 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<i64>>(
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<Vec<i64>, 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<u64, String> {
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<String, Value>,
column: &str,
index: usize,
) -> Result<String, String> {
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<String, Value>, column: &str, index: usize) -> Result<i32, String> {
let value = required_text(row, column, index)?;
value
.parse::<i32>()
.map_err(|_| format!("row {}, column {column} must be an integer", index + 2))
}
fn required_f64(row: &HashMap<String, Value>, column: &str, index: usize) -> Result<f64, String> {
let value = required_text(row, column, index)?;
value
.parse::<f64>()
.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);
}
}

22
apps/api/src/models.rs
View File

@@ -63,6 +63,28 @@ pub struct ImportValidationResponse {
pub errors: Vec<ImportValidationError>,
}
#[derive(Debug, Deserialize)]
pub struct ImportExecutionRequest {
pub template_name: String,
pub run_id: i64,
pub raw_artifact_id: i64,
pub rows: Vec<HashMap<String, Value>>,
pub allow_reimport: Option<bool>,
}
#[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<i64>,
pub status: String,
}
#[derive(Debug, Clone, Serialize, FromRow)]
pub struct AreaScore {
pub area_id: String,

9
apps/api/src/routes.rs
View File

@@ -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))

29
apps/web/src/lib/api.ts
View File

@@ -160,6 +160,26 @@ export type CreateRawArtifact = {
notes?: string;
};
export type ImportExecutionRequest = {
template_name: string;
run_id: number;
raw_artifact_id: number;
rows: Record<string, string | number | null>[];
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<Raw
});
}
export async function executeImport(
payload: ImportExecutionRequest,
): Promise<ImportExecutionResponse> {
return fetchJson("/api/v1/imports/execute", {
method: "POST",
body: JSON.stringify(payload),
});
}
async function fetchJson<T>(path: string, init?: RequestInit): Promise<T> {
const response = await fetch(`${API_BASE_URL}${path}`, {
...init,

38
docs/import_templates/README.md
View File

@@ -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`。

6
docs/product_roadmap.md
View File

@@ -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 资产研究工作台