diff --git a/README.md b/README.md index 81e32ff..7e7eeb2 100644 --- a/README.md +++ b/README.md @@ -60,6 +60,9 @@ 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/raw-artifacts +POST /api/v1/raw-artifacts +POST /api/v1/imports/validate 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/202606240001_raw_artifacts.sql b/apps/api/migrations/202606240001_raw_artifacts.sql new file mode 100644 index 0000000..8898ed6 --- /dev/null +++ b/apps/api/migrations/202606240001_raw_artifacts.sql @@ -0,0 +1,24 @@ +CREATE TABLE IF NOT EXISTS audit.raw_artifacts ( + artifact_id BIGSERIAL PRIMARY KEY, + run_id BIGINT REFERENCES audit.ingestion_runs(run_id), + source_id BIGINT REFERENCES audit.data_sources(source_id), + original_filename TEXT NOT NULL, + raw_uri TEXT NOT NULL, + sha256 TEXT NOT NULL CHECK (sha256 ~ '^[0-9a-f]{64}$'), + size_bytes BIGINT NOT NULL CHECK (size_bytes >= 0), + mime_type TEXT, + uploaded_by TEXT NOT NULL DEFAULT 'system', + notes TEXT NOT NULL DEFAULT '', + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE (sha256) +); + +ALTER TABLE audit.ingestion_runs + ADD COLUMN IF NOT EXISTS raw_artifact_id BIGINT REFERENCES audit.raw_artifacts(artifact_id); + +CREATE INDEX IF NOT EXISTS idx_raw_artifacts_run_id + ON audit.raw_artifacts(run_id); +CREATE INDEX IF NOT EXISTS idx_raw_artifacts_source_id + ON audit.raw_artifacts(source_id); +CREATE INDEX IF NOT EXISTS idx_raw_artifacts_created_at + ON audit.raw_artifacts(created_at DESC); diff --git a/apps/api/src/handlers.rs b/apps/api/src/handlers.rs index ef14fe2..3faa634 100644 --- a/apps/api/src/handlers.rs +++ b/apps/api/src/handlers.rs @@ -5,10 +5,11 @@ use serde::Serialize; use crate::error::{ApiError, ApiResult}; use crate::import_templates::validate_import_payload; use crate::models::{ - AreaScore, AreaScoreSummary, CreateDataSource, CreateIngestionRun, CreateWatchlistItem, - DataSource, FinishIngestionRun, ImportValidationRequest, ImportValidationResponse, - IngestionRun, IngestionRunQuery, MarketOverview, MonthQuery, Neighborhood, NeighborhoodQuery, - NeighborhoodScore, NeighborhoodScoreQuery, UpdateWatchlistItem, WatchlistItem, + AreaScore, AreaScoreSummary, CreateDataSource, CreateIngestionRun, CreateRawArtifact, + CreateWatchlistItem, DataSource, FinishIngestionRun, ImportValidationRequest, + ImportValidationResponse, IngestionRun, IngestionRunQuery, MarketOverview, MonthQuery, + Neighborhood, NeighborhoodQuery, NeighborhoodScore, NeighborhoodScoreQuery, RawArtifact, + RawArtifactQuery, UpdateWatchlistItem, WatchlistItem, }; use crate::state::AppState; @@ -265,6 +266,7 @@ pub async fn list_ingestion_runs( r.run_id, r.source_id, s.name AS source_name, + r.raw_artifact_id, r.status, r.started_at::text AS started_at, r.finished_at::text AS finished_at, @@ -299,10 +301,11 @@ pub async fn create_ingestion_run( source_id, status, raw_uri, + raw_artifact_id, row_count, error_message ) - VALUES ($1, COALESCE($2, 'running'), $3, $4, $5) + VALUES ($1, COALESCE($2, 'running'), $3, $4, $5, $6) RETURNING run_id, source_id, @@ -311,6 +314,7 @@ pub async fn create_ingestion_run( FROM audit.data_sources WHERE source_id = audit.ingestion_runs.source_id ) AS source_name, + raw_artifact_id, status, started_at::text AS started_at, finished_at::text AS finished_at, @@ -322,6 +326,7 @@ pub async fn create_ingestion_run( .bind(payload.source_id) .bind(payload.status) .bind(payload.raw_uri) + .bind(payload.raw_artifact_id) .bind(payload.row_count) .bind(payload.error_message) .fetch_one(&state.pool) @@ -355,6 +360,7 @@ pub async fn finish_ingestion_run( FROM audit.data_sources WHERE source_id = audit.ingestion_runs.source_id ) AS source_name, + raw_artifact_id, status, started_at::text AS started_at, finished_at::text AS finished_at, @@ -374,6 +380,154 @@ pub async fn finish_ingestion_run( Ok(Json(run)) } +pub async fn list_raw_artifacts( + State(state): State, + Query(query): Query, +) -> ApiResult>> { + if let Some(sha256) = query.sha256.as_deref() { + validate_sha256(sha256)?; + } + + let artifacts = sqlx::query_as::<_, RawArtifact>( + r#" + SELECT + a.artifact_id, + a.run_id, + a.source_id, + s.name AS source_name, + a.original_filename, + a.raw_uri, + a.sha256, + a.size_bytes, + a.mime_type, + a.uploaded_by, + a.notes, + a.created_at::text AS created_at + FROM audit.raw_artifacts a + LEFT JOIN audit.data_sources s ON s.source_id = a.source_id + WHERE ($1::bigint IS NULL OR a.source_id = $1) + AND ( + $2::bigint IS NULL + OR a.run_id = $2 + OR EXISTS ( + SELECT 1 + FROM audit.ingestion_runs r + WHERE r.run_id = $2 + AND r.raw_artifact_id = a.artifact_id + ) + ) + AND ($3::text IS NULL OR a.sha256 = $3) + ORDER BY a.created_at DESC, a.artifact_id DESC + LIMIT 100 + "#, + ) + .bind(query.source_id) + .bind(query.run_id) + .bind(query.sha256) + .fetch_all(&state.pool) + .await?; + + Ok(Json(artifacts)) +} + +pub async fn create_raw_artifact( + State(state): State, + Json(payload): Json, +) -> ApiResult> { + payload + .validate() + .map_err(|message| ApiError::BadRequest(message.to_string()))?; + + let mut tx = state.pool.begin().await?; + let artifact = sqlx::query_as::<_, RawArtifact>( + r#" + INSERT INTO audit.raw_artifacts ( + run_id, + source_id, + original_filename, + raw_uri, + sha256, + size_bytes, + mime_type, + uploaded_by, + notes + ) + VALUES ( + $1, + COALESCE( + $2, + ( + SELECT source_id + FROM audit.ingestion_runs + WHERE run_id = $1 + ) + ), + $3, + $4, + $5, + $6, + $7, + COALESCE($8, 'system'), + COALESCE($9, '') + ) + ON CONFLICT (sha256) DO UPDATE + SET run_id = COALESCE(audit.raw_artifacts.run_id, EXCLUDED.run_id), + source_id = COALESCE(audit.raw_artifacts.source_id, EXCLUDED.source_id), + notes = CASE + WHEN audit.raw_artifacts.notes = '' THEN EXCLUDED.notes + ELSE audit.raw_artifacts.notes + END + RETURNING + artifact_id, + run_id, + source_id, + ( + SELECT name + FROM audit.data_sources + WHERE source_id = audit.raw_artifacts.source_id + ) AS source_name, + original_filename, + raw_uri, + sha256, + size_bytes, + mime_type, + uploaded_by, + notes, + created_at::text AS created_at + "#, + ) + .bind(payload.run_id) + .bind(payload.source_id) + .bind(payload.original_filename.trim().to_string()) + .bind(payload.raw_uri.trim().to_string()) + .bind(payload.sha256.trim().to_string()) + .bind(payload.size_bytes) + .bind(payload.mime_type) + .bind(payload.uploaded_by) + .bind(payload.notes) + .fetch_one(&mut *tx) + .await?; + + if let Some(run_id) = payload.run_id { + sqlx::query( + r#" + UPDATE audit.ingestion_runs + SET raw_artifact_id = $2, + raw_uri = COALESCE(raw_uri, $3) + WHERE run_id = $1 + "#, + ) + .bind(run_id) + .bind(artifact.artifact_id) + .bind(&artifact.raw_uri) + .execute(&mut *tx) + .await?; + } + + tx.commit().await?; + Ok(Json(artifact)) +} + pub async fn list_watchlist_items( State(state): State, ) -> ApiResult>> { @@ -563,14 +717,29 @@ fn validate_month(month: &str) -> ApiResult<()> { } } +fn validate_sha256(value: &str) -> ApiResult<()> { + let valid = value.len() == 64 + && value + .chars() + .all(|item| item.is_ascii_hexdigit() && !item.is_ascii_uppercase()); + + if valid { + Ok(()) + } else { + Err(ApiError::BadRequest( + "sha256 must be a lowercase 64-character hex digest".to_string(), + )) + } +} + #[cfg(test)] mod tests { use crate::models::{ - CreateDataSource, CreateIngestionRun, CreateWatchlistItem, FinishIngestionRun, - UpdateWatchlistItem, + CreateDataSource, CreateIngestionRun, CreateRawArtifact, CreateWatchlistItem, + FinishIngestionRun, UpdateWatchlistItem, }; - use super::validate_month; + use super::{validate_month, validate_sha256}; #[test] fn accepts_valid_month() { @@ -674,6 +843,7 @@ mod tests { source_id: Some(1), status: Some("running".to_string()), raw_uri: Some("raw/official/2026-05.csv".to_string()), + raw_artifact_id: None, row_count: Some(10), error_message: None, } @@ -684,6 +854,7 @@ mod tests { source_id: Some(1), status: Some("unknown".to_string()), raw_uri: None, + raw_artifact_id: None, row_count: None, error_message: None, } @@ -706,4 +877,42 @@ mod tests { .validate() .is_err()); } + + #[test] + fn validates_raw_artifact_payload() { + assert!(CreateRawArtifact { + run_id: None, + source_id: None, + original_filename: "area.csv".to_string(), + raw_uri: "raw://sha256/area.csv".to_string(), + sha256: "a".repeat(64), + size_bytes: 128, + mime_type: Some("text/csv".to_string()), + uploaded_by: Some("researcher".to_string()), + notes: None, + } + .validate() + .is_ok()); + + assert!(CreateRawArtifact { + run_id: None, + source_id: None, + original_filename: "area.csv".to_string(), + raw_uri: "raw://sha256/area.csv".to_string(), + sha256: "A".repeat(64), + size_bytes: 128, + mime_type: None, + uploaded_by: None, + notes: None, + } + .validate() + .is_err()); + } + + #[test] + fn validates_sha256_query_value() { + assert!(validate_sha256(&"0".repeat(64)).is_ok()); + assert!(validate_sha256(&"G".repeat(64)).is_err()); + assert!(validate_sha256("short").is_err()); + } } diff --git a/apps/api/src/models.rs b/apps/api/src/models.rs index f0a1897..51c35ab 100644 --- a/apps/api/src/models.rs +++ b/apps/api/src/models.rs @@ -25,6 +25,13 @@ pub struct IngestionRunQuery { pub source_id: Option, } +#[derive(Debug, Deserialize)] +pub struct RawArtifactQuery { + pub source_id: Option, + pub run_id: Option, + pub sha256: Option, +} + #[derive(Debug, Deserialize)] pub struct ImportValidationRequest { pub template_name: String, @@ -211,6 +218,7 @@ pub struct IngestionRun { pub run_id: i64, pub source_id: Option, pub source_name: Option, + pub raw_artifact_id: Option, pub status: String, pub started_at: String, pub finished_at: Option, @@ -224,6 +232,7 @@ pub struct CreateIngestionRun { pub source_id: Option, pub status: Option, pub raw_uri: Option, + pub raw_artifact_id: Option, pub row_count: Option, pub error_message: Option, } @@ -250,6 +259,67 @@ impl CreateIngestionRun { } } +#[derive(Debug, Clone, Serialize, FromRow)] +pub struct RawArtifact { + pub artifact_id: i64, + pub run_id: Option, + pub source_id: Option, + pub source_name: Option, + pub original_filename: String, + pub raw_uri: String, + pub sha256: String, + pub size_bytes: i64, + pub mime_type: Option, + pub uploaded_by: String, + pub notes: String, + pub created_at: String, +} + +#[derive(Debug, Deserialize)] +pub struct CreateRawArtifact { + pub run_id: Option, + pub source_id: Option, + pub original_filename: String, + pub raw_uri: String, + pub sha256: String, + pub size_bytes: i64, + pub mime_type: Option, + pub uploaded_by: Option, + pub notes: Option, +} + +impl CreateRawArtifact { + pub fn validate(&self) -> Result<(), &'static str> { + if self.original_filename.trim().is_empty() { + return Err("original_filename is required"); + } + if self.raw_uri.trim().is_empty() { + return Err("raw_uri is required"); + } + if !is_valid_sha256(&self.sha256) { + return Err("sha256 must be a lowercase 64-character hex digest"); + } + if self.size_bytes < 0 { + return Err("size_bytes must be non-negative"); + } + for value in [ + Some(self.original_filename.as_str()), + Some(self.raw_uri.as_str()), + self.mime_type.as_deref(), + self.uploaded_by.as_deref(), + self.notes.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, @@ -319,6 +389,13 @@ fn has_sensitive_text(value: &str) -> bool { || value.starts_with("/private/") } +fn is_valid_sha256(value: &str) -> bool { + value.len() == 64 + && value + .chars() + .all(|item| item.is_ascii_hexdigit() && !item.is_ascii_uppercase()) +} + #[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 5c811ad..c0d3c1d 100644 --- a/apps/api/src/routes.rs +++ b/apps/api/src/routes.rs @@ -4,10 +4,11 @@ use tower_http::cors::CorsLayer; use tower_http::trace::TraceLayer; use crate::handlers::{ - 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, validate_import, + 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, }; use crate::state::AppState; @@ -31,6 +32,10 @@ pub fn build_router(state: AppState) -> Router { "/ingestion-runs/{run_id}/finish", patch(finish_ingestion_run), ) + .route( + "/raw-artifacts", + get(list_raw_artifacts).post(create_raw_artifact), + ) .route("/imports/validate", axum::routing::post(validate_import)) .route("/market/overview", get(market_overview)) .route("/neighborhoods", get(list_neighborhoods)) diff --git a/apps/web/src/components/dashboard/market-dashboard.tsx b/apps/web/src/components/dashboard/market-dashboard.tsx index 3cbd964..4e11c36 100644 --- a/apps/web/src/components/dashboard/market-dashboard.tsx +++ b/apps/web/src/components/dashboard/market-dashboard.tsx @@ -27,12 +27,14 @@ import { archiveWatchlistItem, createDataSource, createIngestionRun, + createRawArtifact, createWatchlistItem, fetchDataSources, fetchAreaScores, fetchIngestionRuns, fetchMarketOverview, fetchNeighborhoodScores, + fetchRawArtifacts, fetchWatchlistItems, finishIngestionRun, } from "@/lib/api"; @@ -40,10 +42,12 @@ import type { AreaScore, CreateDataSource, CreateIngestionRun, + CreateRawArtifact, CreateWatchlistItem, DataSource, IngestionRun, NeighborhoodScore, + RawArtifact, WatchlistItem, } from "@/lib/api"; import { formatNumber, formatPrice } from "@/lib/utils"; @@ -88,6 +92,10 @@ export function MarketDashboard() { queryKey: ["ingestion-runs"], queryFn: fetchIngestionRuns, }); + const rawArtifacts = useQuery({ + queryKey: ["raw-artifacts"], + queryFn: fetchRawArtifacts, + }); const createMutation = useMutation({ mutationFn: createWatchlistItem, onSuccess: async () => { @@ -112,6 +120,15 @@ export function MarketDashboard() { await queryClient.invalidateQueries({ queryKey: ["ingestion-runs"] }); }, }); + const createRawArtifactMutation = useMutation({ + mutationFn: createRawArtifact, + onSuccess: async () => { + await Promise.all([ + queryClient.invalidateQueries({ queryKey: ["raw-artifacts"] }), + queryClient.invalidateQueries({ queryKey: ["ingestion-runs"] }), + ]); + }, + }); const finishRunMutation = useMutation({ mutationFn: ({ id, rowCount }: { id: number; rowCount?: number }) => finishIngestionRun(id, { status: "succeeded", row_count: rowCount }), @@ -127,13 +144,15 @@ export function MarketDashboard() { watchlist.isLoading || dataSources.isLoading || ingestionRuns.isLoading; + const importLoading = ingestionRuns.isLoading || rawArtifacts.isLoading; const hasError = overview.isError || scores.isError || neighborhoodScores.isError || watchlist.isError || dataSources.isError || - ingestionRuns.isError; + ingestionRuns.isError || + rawArtifacts.isError; return (
@@ -156,6 +175,7 @@ export function MarketDashboard() { void watchlist.refetch(); void dataSources.refetch(); void ingestionRuns.refetch(); + void rawArtifacts.refetch(); }} disabled={isLoading} aria-label="刷新" @@ -276,12 +296,16 @@ export function MarketDashboard() { createDataSourceMutation.mutate(payload)} onCreateRun={(payload) => createRunMutation.mutate(payload)} + onCreateArtifact={(payload) => createRawArtifactMutation.mutate(payload)} onFinishRun={(id, rowCount) => finishRunMutation.mutate({ id, rowCount })} /> @@ -292,22 +316,30 @@ export function MarketDashboard() { function ImportCenterPanel({ sources, runs, + artifacts, loading, + importLoading, creatingSource, creatingRun, + creatingArtifact, finishingRunId, onCreateSource, onCreateRun, + onCreateArtifact, onFinishRun, }: { sources: DataSource[]; runs: IngestionRun[]; + artifacts: RawArtifact[]; loading: boolean; + importLoading: boolean; creatingSource: boolean; creatingRun: boolean; + creatingArtifact: boolean; finishingRunId: number | null; onCreateSource: (payload: CreateDataSource) => void; onCreateRun: (payload: CreateIngestionRun) => void; + onCreateArtifact: (payload: CreateRawArtifact) => void; onFinishRun: (id: number, rowCount?: number) => void; }) { const [sourceName, setSourceName] = useState(""); @@ -316,6 +348,12 @@ function ImportCenterPanel({ const [sourceId, setSourceId] = useState(""); const [rawUri, setRawUri] = useState(""); const [rowCount, setRowCount] = useState(""); + const [artifactRunId, setArtifactRunId] = useState(""); + const [artifactSourceId, setArtifactSourceId] = useState(""); + const [artifactFilename, setArtifactFilename] = useState(""); + const [artifactRawUri, setArtifactRawUri] = useState(""); + const [artifactSha256, setArtifactSha256] = useState(""); + const [artifactSizeBytes, setArtifactSizeBytes] = useState(""); function submitSource(event: React.FormEvent) { event.preventDefault(); @@ -348,6 +386,36 @@ function ImportCenterPanel({ setRawUri(""); } + function submitArtifact(event: React.FormEvent) { + event.preventDefault(); + const parsedRunId = Number(artifactRunId); + const parsedSourceId = Number(artifactSourceId); + const parsedSizeBytes = Number(artifactSizeBytes); + if ( + !artifactFilename.trim() || + !artifactRawUri.trim() || + artifactSha256.trim().length !== 64 || + !Number.isFinite(parsedSizeBytes) || + parsedSizeBytes < 0 + ) { + return; + } + onCreateArtifact({ + run_id: Number.isFinite(parsedRunId) && parsedRunId > 0 ? parsedRunId : undefined, + source_id: + Number.isFinite(parsedSourceId) && parsedSourceId > 0 ? parsedSourceId : undefined, + original_filename: artifactFilename.trim(), + raw_uri: artifactRawUri.trim(), + sha256: artifactSha256.trim(), + size_bytes: parsedSizeBytes, + uploaded_by: "dashboard", + }); + setArtifactFilename(""); + setArtifactRawUri(""); + setArtifactSha256(""); + setArtifactSizeBytes(""); + } + return ( @@ -415,7 +483,7 @@ function ImportCenterPanel({ +
+
+ + + setArtifactFilename(event.target.value)} + /> + setArtifactRawUri(event.target.value)} + /> + setArtifactSha256(event.target.value)} + /> + setArtifactSizeBytes(event.target.value)} + /> + +
+ +
); @@ -539,6 +674,47 @@ function IngestionRunTable({ ); } +function RawArtifactTable({ + artifacts, + loading, +}: { + artifacts: RawArtifact[]; + loading: boolean; +}) { + if (loading) { + return
; + } + + return ( +
+ + + + 原始文件 + 数据源 + 批次 + SHA-256 + 大小 + + + + {artifacts.map((artifact) => ( + + {artifact.original_filename} + {artifact.source_name ?? "-"} + {artifact.run_id ? `#${artifact.run_id}` : "-"} + + {artifact.sha256} + + {formatNumber(artifact.size_bytes)} + + ))} + +
+
+ ); +} + function NeighborhoodRankingPanel({ scores, neighborhoodScores, diff --git a/apps/web/src/lib/api.ts b/apps/web/src/lib/api.ts index 99567fa..a45678b 100644 --- a/apps/web/src/lib/api.ts +++ b/apps/web/src/lib/api.ts @@ -109,6 +109,7 @@ export type IngestionRun = { run_id: number; source_id: number | null; source_name: string | null; + raw_artifact_id: number | null; status: string; started_at: string; finished_at: string | null; @@ -121,6 +122,7 @@ export type CreateIngestionRun = { source_id?: number; status?: string; raw_uri?: string; + raw_artifact_id?: number; row_count?: number; error_message?: string; }; @@ -131,6 +133,33 @@ export type FinishIngestionRun = { error_message?: string; }; +export type RawArtifact = { + artifact_id: number; + run_id: number | null; + source_id: number | null; + source_name: string | null; + original_filename: string; + raw_uri: string; + sha256: string; + size_bytes: number; + mime_type: string | null; + uploaded_by: string; + notes: string; + created_at: string; +}; + +export type CreateRawArtifact = { + run_id?: number; + source_id?: number; + original_filename: string; + raw_uri: string; + sha256: string; + size_bytes: number; + mime_type?: string; + uploaded_by?: string; + notes?: string; +}; + const API_BASE_URL = process.env.NEXT_PUBLIC_API_BASE_URL?.replace(/\/$/, "") ?? "http://127.0.0.1:8080"; @@ -197,6 +226,17 @@ export async function finishIngestionRun( }); } +export async function fetchRawArtifacts(): Promise { + return fetchJson("/api/v1/raw-artifacts"); +} + +export async function createRawArtifact(payload: CreateRawArtifact): Promise { + return fetchJson("/api/v1/raw-artifacts", { + 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 fd67ba7..b682bf1 100644 --- a/docs/import_templates/README.md +++ b/docs/import_templates/README.md @@ -46,6 +46,17 @@ PYTHONPATH=src python3 -m shanghai_housing validate-import \ 通过时输出行数和模板名；失败时逐行返回错误信息。 +## 原始文件指纹 + +导入前先计算文件指纹，用于登记 `audit.raw_artifacts` 并识别重复文件。 + +```bash +PYTHONPATH=src python3 -m shanghai_housing fingerprint-file \ + --file docs/import_templates/area_monthly_metrics.csv +``` + +输出内容包含 `original_filename`、`raw_uri`、`sha256`、`size_bytes` 和 `mime_type`，可作为 `POST /api/v1/raw-artifacts` 的主要 payload。 + ## API 校验 ```bash @@ -72,3 +83,22 @@ curl -X POST http://127.0.0.1:8080/api/v1/imports/validate \ ] }' ``` + +## API 原始文件登记 + +```bash +curl -X POST http://127.0.0.1:8080/api/v1/raw-artifacts \ + -H 'content-type: application/json' \ + -d '{ + "run_id": 1, + "source_id": 1, + "original_filename": "area_monthly_metrics.csv", + "raw_uri": "raw://sha256//area_monthly_metrics.csv", + "sha256": "<64-character-lowercase-sha256>", + "size_bytes": 2048, + "mime_type": "text/csv", + "uploaded_by": "researcher" + }' +``` + +同一个 `sha256` 重复登记时，API 返回已有 artifact，并可将新的导入批次关联到该 artifact。 diff --git a/docs/product_roadmap.md b/docs/product_roadmap.md index 19d5255..b8a44bb 100644 --- a/docs/product_roadmap.md +++ b/docs/product_roadmap.md @@ -100,22 +100,45 @@ ### M1.4 原始数据留存与文件哈希 +状态：已完成。 + 目标： - 每次导入都记录原始文件位置、哈希、大小、导入人和入库批次。 -- 正式导入结果写入 `raw`/`silver`/`audit`，并与 `audit.ingestion_runs` 关联。 +- 将原始文件资产与 `audit.ingestion_runs` 关联，支持重复文件识别。 交付物： - `audit.raw_artifacts` - 导入批次关联 raw artifact。 -- PostgreSQL 导入执行接口。 +- `GET /api/v1/raw-artifacts` +- `POST /api/v1/raw-artifacts` +- Python 文件指纹命令。 验收标准： - 同一文件重复导入可识别。 - 任一指标可以追溯到原始数据批次。 +### M1.5 PostgreSQL 导入执行接口 + +目标： + +- 将已校验模板数据正式写入 PostgreSQL 的 `raw`/`silver`/`audit`。 +- 导入执行必须关联 `audit.ingestion_runs` 和 `audit.raw_artifacts`。 + +交付物： + +- PostgreSQL 导入执行 API。 +- 支持板块月度指标和小区月度指标的 upsert。 +- 导入失败时写入明确错误并保留审计记录。 + +验收标准： + +- 校验失败不写入业务表。 +- 导入成功后可通过批次追溯到原始文件哈希。 +- 重复导入同一文件时能识别并提示。 + ## M2 资产研究工作台 ### M2.1 板块详情页 diff --git a/src/shanghai_housing/cli.py b/src/shanghai_housing/cli.py index 88938f6..a5c7099 100644 --- a/src/shanghai_housing/cli.py +++ b/src/shanghai_housing/cli.py @@ -7,6 +7,7 @@ from typing import Optional from .db import DEFAULT_DB_PATH, connect, import_template_file, init_db, load_sample_data from .indicators import compute_area_scores from .import_templates import ImportTemplateError, validate_tabular_file +from .raw_artifacts import fingerprint_file from .reporting import format_scores_table, write_monthly_report @@ -69,6 +70,17 @@ def build_parser() -> argparse.ArgumentParser: help="CSV, XLSX, or XLSM file to import.", ) + fingerprint_parser = subparsers.add_parser( + "fingerprint-file", + help="Compute raw artifact metadata for an import file.", + ) + fingerprint_parser.add_argument( + "--file", + type=Path, + required=True, + help="File to hash and describe.", + ) + return parser @@ -132,5 +144,10 @@ def main(argv: Optional[list[str]] = None) -> int: ) return 0 + if args.command == "fingerprint-file": + fingerprint = fingerprint_file(args.file) + print(fingerprint.to_json()) + return 0 + parser.error(f"Unknown command: {args.command}") return 2 diff --git a/src/shanghai_housing/raw_artifacts.py b/src/shanghai_housing/raw_artifacts.py new file mode 100644 index 0000000..4732997 --- /dev/null +++ b/src/shanghai_housing/raw_artifacts.py @@ -0,0 +1,40 @@ +from __future__ import annotations + +import hashlib +import json +import mimetypes +from dataclasses import asdict, dataclass +from pathlib import Path + + +@dataclass(frozen=True) +class RawArtifactFingerprint: + original_filename: str + raw_uri: str + sha256: str + size_bytes: int + mime_type: str | None + + def to_json(self) -> str: + return json.dumps(asdict(self), ensure_ascii=False, indent=2) + + +def fingerprint_file(file_path: Path) -> RawArtifactFingerprint: + path = file_path.resolve() + digest = hashlib.sha256() + size_bytes = 0 + + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + size_bytes += len(chunk) + + sha256 = digest.hexdigest() + mime_type, _ = mimetypes.guess_type(path.name) + return RawArtifactFingerprint( + original_filename=path.name, + raw_uri=f"raw://sha256/{sha256}/{path.name}", + sha256=sha256, + size_bytes=size_bytes, + mime_type=mime_type, + ) diff --git a/tests/test_import_templates.py b/tests/test_import_templates.py index 734905e..b26f43b 100644 --- a/tests/test_import_templates.py +++ b/tests/test_import_templates.py @@ -7,6 +7,7 @@ from pathlib import Path from shanghai_housing.db import connect, import_template_file from shanghai_housing.import_templates import validate_csv_file, validate_import_rows +from shanghai_housing.raw_artifacts import fingerprint_file class ImportTemplateTests(unittest.TestCase): @@ -79,6 +80,15 @@ class ImportTemplateTests(unittest.TestCase): self.assertEqual(len(rows), 1) self.assertEqual(rows[0]["level"], "shanghai") + def test_fingerprints_import_file_without_local_path(self) -> None: + fingerprint = fingerprint_file(Path("docs/import_templates/policy_events.csv")) + + self.assertEqual(fingerprint.original_filename, "policy_events.csv") + self.assertEqual(len(fingerprint.sha256), 64) + self.assertGreater(fingerprint.size_bytes, 0) + self.assertTrue(fingerprint.raw_uri.startswith("raw://sha256/")) + self.assertNotIn("/Users/", fingerprint.raw_uri) + if __name__ == "__main__": unittest.main()