Initial commit for Hugging Face Space

Dockerfile

@@ -0,0 +1,22 @@
+FROM python:3.11-slim
+
+WORKDIR /app
+
+# 安装必要的系统依赖（用于编译 pg 驱动等）
+RUN apt-get update && apt-get install -y gcc libpq-dev && rm -rf /var/lib/apt/lists/*
+
+# 安装 Python 依赖
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# 复制代码
+COPY . .
+
+# 暴露端口
+EXPOSE 8000
+
+# 设置环境变量，使用容器内的数据库地址
+ENV DATABASE_URL="postgresql+asyncpg://postgres:postgres@db:5432/customs_data"
+
+# 启动命令：先运行 mock 脚本初始化库和数据，再运行巴西的测试数据，最后启动 API 服务
+CMD ["bash", "-c", "python apps/worker/run_mock.py && python apps/worker/run_brazil.py && uvicorn apps.api.main:app --host 0.0.0.0 --port 8000"]

apps/api/main.py

@@ -0,0 +1,45 @@
+from contextlib import asynccontextmanager
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.staticfiles import StaticFiles
+from fastapi.responses import FileResponse
+from packages.core.logger import app_logger
+
+# 我们将之前写好的 trade 接口引入
+from apps.api.routers import trade
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    # 启动时初始化
+    app_logger.info("Initializing API Service...")
+    yield
+    # 关闭时
+    app_logger.info("Closing API Service...")
+
+app = FastAPI(
+    title="Customs Data API",
+    description="海关数据查询服务 API",
+    version="1.0.0",
+    lifespan=lifespan
+)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# 挂载静态文件
+app.mount("/static", StaticFiles(directory="apps/api/static"), name="static")
+
+@app.get("/", tags=["UI"])
+async def serve_ui():
+    return FileResponse("apps/api/static/index.html")
+
+@app.get("/api/v1/health", tags=["System"])
+async def health_check():
+    return {"status": "ok", "service": "Customs Data API"}
+
+app.include_router(trade.router, prefix="/api/v1/trade", tags=["Trade Search"])

apps/api/routers/trade.py

@@ -0,0 +1,55 @@
+from fastapi import APIRouter, Depends, Query, HTTPException
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select, func, desc
+
+from packages.core.database import get_db_session
+from packages.core.models import StandardTradeRecord
+from apps.api.schemas.trade import TradeQueryRequest, TradeRecordResponse, PaginatedResponse
+
+router = APIRouter()
+
+@router.post("/search", response_model=PaginatedResponse[TradeRecordResponse])
+async def search_trade_records(
+    query: TradeQueryRequest,
+    db: AsyncSession = Depends(get_db_session)
+):
+    """
+    检索标准贸易记录
+    支持按国家、HS 编码、进口商模糊查询及日期范围筛选
+    """
+    stmt = select(StandardTradeRecord)
+    count_stmt = select(func.count()).select_from(StandardTradeRecord)
+    
+    # 动态构建查询条件
+    filters = []
+    if query.source_country:
+        filters.append(StandardTradeRecord.source_country == query.source_country)
+    if query.hs_code:
+        filters.append(StandardTradeRecord.hs_code.like(f"{query.hs_code}%"))
+    if query.importer_name:
+        filters.append(StandardTradeRecord.importer_name.ilike(f"%{query.importer_name}%"))
+    if query.start_date:
+        filters.append(StandardTradeRecord.trade_date >= query.start_date)
+    if query.end_date:
+        filters.append(StandardTradeRecord.trade_date <= query.end_date)
+        
+    if filters:
+        stmt = stmt.where(*filters)
+        count_stmt = count_stmt.where(*filters)
+        
+    # 计算总数
+    count_stmt = select(func.count()).select_from(stmt.subquery())
+    total_result = await db.execute(count_stmt)
+    total = total_result.scalar() or 0
+    
+    # 分页查询
+    stmt = stmt.order_by(desc(StandardTradeRecord.trade_date))
+    stmt = stmt.offset((query.page - 1) * query.limit).limit(query.limit)
+    
+    result = await db.execute(stmt)
+    records = result.scalars().all()
+    
+    return PaginatedResponse(
+        total=total,
+        items=records
+    )

apps/api/schemas/trade.py

@@ -0,0 +1,46 @@
+from pydantic import BaseModel, Field
+from typing import List, Optional, TypeVar, Generic
+from datetime import datetime
+
+class TradeQueryRequest(BaseModel):
+    source_country: Optional[str] = Field(None, description="来源国 (如 US, BR)")
+    hs_code: Optional[str] = Field(None, description="HS编码 (模糊前缀匹配)")
+    importer_name: Optional[str] = Field(None, description="进口商名称 (模糊匹配)")
+    exporter_name: Optional[str] = Field(None, description="出口商名称 (模糊匹配)")
+    start_date: Optional[datetime] = Field(None, description="起始时间")
+    end_date: Optional[datetime] = Field(None, description="结束时间")
+    page: int = Field(1, ge=1, description="页码")
+    limit: int = Field(20, ge=1, le=100, description="每页数量")
+
+    @property
+    def offset(self) -> int:
+        return (self.page - 1) * self.limit
+
+class TradeRecordResponse(BaseModel):
+    record_id: str
+    source_record_id: str
+    source_country: str
+    trade_direction: str
+    trade_date: datetime
+    importer_name: Optional[str]
+    exporter_name: Optional[str]
+    hs_code: Optional[str]
+    product_name: Optional[str]
+    amount: Optional[float]
+    currency: Optional[str]
+    weight: Optional[float]
+    weight_unit: Optional[str]
+    origin_country: Optional[str]
+    destination_country: Optional[str]
+    departure_port: Optional[str]
+    arrival_port: Optional[str]
+    transport_mode: Optional[str]
+
+    class Config:
+        from_attributes = True
+
+T = TypeVar('T')
+
+class PaginatedResponse(BaseModel, Generic[T]):
+    total: int
+    items: List[T]

apps/api/static/index.html

@@ -0,0 +1,219 @@
+<!DOCTYPE html>
+<html>
+<head>
+    <meta charset="UTF-8">
+    <title>海关数据查询系统 - MVP</title>
+    <link rel="stylesheet" href="https://unpkg.com/element-ui/lib/theme-chalk/index.css">
+    <script src="https://cdn.jsdelivr.net/npm/vue@2.6.14/dist/vue.js"></script>
+    <script src="https://unpkg.com/element-ui/lib/index.js"></script>
+    <script src="https://cdn.jsdelivr.net/npm/axios/dist/axios.min.js"></script>
+    <style>
+        body { margin: 0; padding: 20px; font-family: "Helvetica Neue", Helvetica, "PingFang SC", "Hiragino Sans GB", "Microsoft YaHei", "微软雅黑", Arial, sans-serif; background-color: #f5f7fa; }
+        .app-container { max-width: 1400px; margin: 0 auto; }
+        .header { display: flex; justify-content: space-between; align-items: center; margin-bottom: 20px; }
+        .header h1 { margin: 0; color: #303133; font-size: 24px; }
+        .search-card { margin-bottom: 20px; border-radius: 8px; box-shadow: 0 2px 12px 0 rgba(0,0,0,.1); }
+        .result-card { border-radius: 8px; box-shadow: 0 2px 12px 0 rgba(0,0,0,.1); }
+        .el-table th { background-color: #f5f7fa; color: #606266; }
+        .pagination-container { margin-top: 20px; text-align: right; }
+        .tag-country { font-weight: bold; }
+        .amount-text { color: #f56c6c; font-weight: bold; }
+    </style>
+</head>
+<body>
+    <div id="app" class="app-container">
+        <div class="header">
+            <h1>🌐 海关数据查询系统 - 实时看板</h1>
+            <el-tag type="success">实时数据抓取运行中</el-tag>
+        </div>
+
+        <!-- 搜索条件区 -->
+        <el-card class="search-card">
+            <el-form :inline="true" :model="searchForm" size="small" @submit.native.prevent="handleSearch">
+                <el-form-item label="国家/地区">
+                    <el-select v-model="searchForm.source_country" placeholder="选择国家" clearable style="width: 120px;">
+                        <el-option label="全部" value=""></el-option>
+                        <el-option label="美国 (US)" value="US"></el-option>
+                        <el-option label="巴西 (BR)" value="BR"></el-option>
+                        <el-option label="印度尼西亚 (ID)" value="ID"></el-option>
+                        <el-option label="泰国 (TH)" value="TH"></el-option>
+                        <el-option label="越南 (VN)" value="VN"></el-option>
+                        <el-option label="马来西亚 (MY)" value="MY"></el-option>
+                        <el-option label="新西兰 (NZ)" value="NZ"></el-option>
+                        <el-option label="阿联酋 (AE)" value="AE"></el-option>
+                        <el-option label="德国 (DE)" value="DE"></el-option>
+                    </el-select>
+                </el-form-item>
+                <el-form-item label="进口商">
+                    <el-input v-model="searchForm.importer_name" placeholder="支持模糊搜索" clearable></el-input>
+                </el-form-item>
+                <el-form-item label="出口商">
+                    <el-input v-model="searchForm.exporter_name" placeholder="支持模糊搜索" clearable></el-input>
+                </el-form-item>
+                <el-form-item label="HS编码">
+                    <el-input v-model="searchForm.hs_code" placeholder="如: 8471" clearable style="width: 120px;"></el-input>
+                </el-form-item>
+                <el-form-item>
+                    <el-button type="primary" icon="el-icon-search" @click="handleSearch" :loading="loading">查询</el-button>
+                    <el-button icon="el-icon-refresh" @click="resetSearch">重置</el-button>
+                </el-form-item>
+            </el-form>
+        </el-card>
+
+        <!-- 数据表格区 -->
+        <el-card class="result-card">
+            <div slot="header" class="clearfix">
+                <span>共找到 <b>{{ total }}</b> 条清洗后的标准提单记录</span>
+                <el-button style="float: right; padding: 3px 0" type="text" icon="el-icon-download">导出当前数据</el-button>
+            </div>
+            
+            <el-table :data="tableData" v-loading="loading" style="width: 100%" border size="small" stripe>
+                <el-table-column prop="trade_date" label="交易日期" width="100" align="center">
+                    <template slot-scope="scope">{{ formatDate(scope.row.trade_date) }}</template>
+                </el-table-column>
+                
+                <el-table-column label="方向/国家" width="110" align="center">
+                    <template slot-scope="scope">
+                        <el-tag :type="scope.row.trade_direction === 'import' ? 'primary' : 'warning'" size="mini" effect="dark" style="margin-bottom: 4px;">
+                            {{ scope.row.trade_direction === 'import' ? '进口' : '出口' }}
+                        </el-tag><br>
+                        <span class="tag-country">{{ scope.row.source_country }}</span>
+                    </template>
+                </el-table-column>
+                
+                <el-table-column label="交易双方" min-width="220">
+                    <template slot-scope="scope">
+                        <div style="font-size: 12px; color: #909399;">进口商：</div>
+                        <div style="font-weight: bold; margin-bottom: 8px;">{{ scope.row.importer_name || '-' }}</div>
+                        <div style="font-size: 12px; color: #909399;">出口商：</div>
+                        <div style="font-weight: bold;">{{ scope.row.exporter_name || '-' }}</div>
+                    </template>
+                </el-table-column>
+
+                <el-table-column label="商品信息" min-width="200">
+                    <template slot-scope="scope">
+                        <el-tag size="mini" type="info" style="margin-bottom: 4px;">HS: {{ scope.row.hs_code || '-' }}</el-tag>
+                        <div style="font-size: 12px; line-height: 1.4; color: #606266;">
+                            {{ scope.row.product_name || '-' }}
+                        </div>
+                    </template>
+                </el-table-column>
+
+                <el-table-column label="金额/重量" width="160" align="right">
+                    <template slot-scope="scope">
+                        <div class="amount-text" v-if="scope.row.amount">
+                            {{ formatNumber(scope.row.amount) }} {{ scope.row.currency }}
+                        </div>
+                        <div v-else>-</div>
+                        <div style="font-size: 12px; color: #909399; margin-top: 4px;" v-if="scope.row.weight">
+                            {{ formatNumber(scope.row.weight) }} {{ scope.row.weight_unit }}
+                        </div>
+                    </template>
+                </el-table-column>
+
+                <el-table-column label="航线信息" width="160">
+                    <template slot-scope="scope">
+                        <div style="font-size: 12px;"><i class="el-icon-location-outline"></i> 原产: {{ scope.row.origin_country || '-' }}</div>
+                        <div style="font-size: 12px;"><i class="el-icon-place"></i> 目的: {{ scope.row.destination_country || '-' }}</div>
+                        <div style="font-size: 12px; margin-top: 4px; color: #909399;">
+                            <i class="el-icon-ship"></i> 方式: {{ scope.row.transport_mode || '-' }}
+                        </div>
+                    </template>
+                </el-table-column>
+            </el-table>
+
+            <div class="pagination-container">
+                <el-pagination
+                    background
+                    @size-change="handleSizeChange"
+                    @current-change="handleCurrentChange"
+                    :current-page="searchForm.page"
+                    :page-sizes="[10, 20, 50, 100]"
+                    :page-size="searchForm.limit"
+                    layout="total, sizes, prev, pager, next, jumper"
+                    :total="total">
+                </el-pagination>
+            </div>
+        </el-card>
+    </div>
+
+    <script>
+        new Vue({
+            el: '#app',
+            data: function() {
+                return {
+                    loading: false,
+                    tableData: [],
+                    total: 0,
+                    searchForm: {
+                        source_country: '',
+                        importer_name: '',
+                        exporter_name: '',
+                        hs_code: '',
+                        page: 1,
+                        limit: 10
+                    }
+                }
+            },
+            mounted() {
+                this.fetchData();
+            },
+            methods: {
+                async fetchData() {
+                    this.loading = true;
+                    try {
+                        // 清理空参数
+                        const payload = {};
+                        for (const key in this.searchForm) {
+                            if (this.searchForm[key] !== '' && this.searchForm[key] !== null) {
+                                payload[key] = this.searchForm[key];
+                            }
+                        }
+                        
+                        const response = await axios.post('/api/v1/trade/search', payload);
+                        this.tableData = response.data.items;
+                        this.total = response.data.total;
+                    } catch (error) {
+                        this.$message.error('获取数据失败，请检查网络或刷新重试');
+                        console.error(error);
+                    } finally {
+                        this.loading = false;
+                    }
+                },
+                handleSearch() {
+                    this.searchForm.page = 1;
+                    this.fetchData();
+                },
+                resetSearch() {
+                    this.searchForm = {
+                        source_country: '',
+                        importer_name: '',
+                        exporter_name: '',
+                        hs_code: '',
+                        page: 1,
+                        limit: 10
+                    };
+                    this.fetchData();
+                },
+                handleSizeChange(val) {
+                    this.searchForm.limit = val;
+                    this.fetchData();
+                },
+                handleCurrentChange(val) {
+                    this.searchForm.page = val;
+                    this.fetchData();
+                },
+                formatDate(dateString) {
+                    if (!dateString) return '-';
+                    const date = new Date(dateString);
+                    return `${date.getFullYear()}-${String(date.getMonth() + 1).padStart(2, '0')}-${String(date.getDate()).padStart(2, '0')}`;
+                },
+                formatNumber(num) {
+                    if (num === null || num === undefined) return '-';
+                    return Number(num).toLocaleString('en-US');
+                }
+            }
+        })
+    </script>
+</body>
+</html>

apps/worker/run_backfill.py

@@ -0,0 +1,70 @@
+import asyncio
+import sys
+import os
+import argparse
+from datetime import datetime, timedelta
+
+sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
+
+from packages.core.logger import app_logger
+
+async def run_backfill(country_code: str, start_date: str, end_date: str):
+    """
+    历史数据回溯脚本 (Backfill)
+    允许指定国家和时间范围，批量补录过去的数据。
+    """
+    app_logger.info(f"--- Starting BACKFILL Job for {country_code} from {start_date} to {end_date} ---")
+    
+    # 如果环境变量中没有 DATABASE_URL，则设置一个默认值（兼容宿主机直接运行）
+    if "DATABASE_URL" not in os.environ:
+        os.environ["DATABASE_URL"] = "postgresql+asyncpg://postgres:postgres@localhost:5433/customs_data"
+    
+    from packages.core.database import AsyncSessionLocal
+    from packages.connectors.mock.us_mock import MockUSConnector
+    from packages.connectors.brazil import BrazilComexStatConnector
+    from packages.connectors.mock.extended_mock import ExtendedMockConnector, COUNTRY_CONFIGS
+    
+    start_dt = datetime.strptime(start_date, "%Y-%m-%d")
+    end_dt = datetime.strptime(end_date, "%Y-%m-%d")
+    
+    async with AsyncSessionLocal() as session:
+        # 1. 实例化对应的 Connector
+        if country_code == "US":
+            connector = MockUSConnector(session=session)
+        elif country_code == "BR":
+            connector = BrazilComexStatConnector(session=session)
+        elif country_code in COUNTRY_CONFIGS:
+            connector = ExtendedMockConnector(country_code=country_code, session=session)
+        else:
+            app_logger.error(f"Unsupported country code for backfill: {country_code}")
+            return
+
+        # 2. 模拟按月拆分任务进行回溯
+        # 真实场景下，海关接口通常限制单次查询跨度，比如只能查一个月
+        current_dt = start_dt
+        while current_dt <= end_dt:
+            next_month = current_dt.replace(day=28) + timedelta(days=4)
+            next_month_start = next_month.replace(day=1)
+            batch_end_dt = min(next_month_start - timedelta(days=1), end_dt)
+            
+            app_logger.info(f"[{country_code}] Backfilling for period: {current_dt.strftime('%Y-%m-%d')} to {batch_end_dt.strftime('%Y-%m-%d')}")
+            
+            # TODO: 真实场景下，这里要将 current_dt 和 batch_end_dt 传给 fetch/discover
+            # 目前 MVP 阶段调用 run() 模拟执行一轮抓取
+            # 为了防止冲突，强行注入一个特定的 batch_no 前缀
+            connector.batch_no = f"BACKFILL_{country_code}_{current_dt.strftime('%Y%m')}"
+            await connector.run()
+            
+            current_dt = next_month_start
+            
+    app_logger.info(f"--- Finished BACKFILL Job for {country_code} ---")
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Customs Data Backfill Script")
+    parser.add_argument("--country", type=str, required=True, help="Country Code (e.g., US, BR, ID)")
+    parser.add_argument("--start", type=str, required=True, help="Start Date (YYYY-MM-DD)")
+    parser.add_argument("--end", type=str, required=True, help="End Date (YYYY-MM-DD)")
+    
+    args = parser.parse_args()
+    
+    asyncio.run(run_backfill(country_code=args.country, start_date=args.start, end_date=args.end))

apps/worker/run_brazil.py

@@ -0,0 +1,21 @@
+import asyncio
+import sys
+import os
+
+# 将项目根目录加入 sys.path
+sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
+
+from packages.core.logger import app_logger
+from packages.core.database import AsyncSessionLocal
+from packages.connectors.brazil import BrazilComexStatConnector
+
+async def run_brazil_job():
+    """运行巴西海关数据采集链路"""
+    app_logger.info("Starting Brazil data sync job...")
+    
+    async with AsyncSessionLocal() as session:
+        connector = BrazilComexStatConnector(session)
+        await connector.run()
+
+if __name__ == "__main__":
+    asyncio.run(run_brazil_job())

apps/worker/run_extended.py

@@ -0,0 +1,24 @@
+import asyncio
+import sys
+import os
+
+sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
+
+from packages.core.logger import app_logger
+from packages.connectors.mock.extended_mock import ExtendedMockConnector, COUNTRY_CONFIGS
+
+from packages.core.database import AsyncSessionLocal
+
+async def run_extended_mock_jobs():
+    """
+    循环跑批处理 7 个新国家的采集任务
+    """
+    async with AsyncSessionLocal() as session:
+        for country_code in COUNTRY_CONFIGS.keys():
+            app_logger.info(f"--- Starting Sync Job for {country_code} ---")
+            connector = ExtendedMockConnector(country_code=country_code, session=session)
+            await connector.run()
+            app_logger.info(f"--- Finished Sync Job for {country_code} ---")
+
+if __name__ == "__main__":
+    asyncio.run(run_extended_mock_jobs())

apps/worker/run_mock.py

@@ -0,0 +1,35 @@
+import asyncio
+import sys
+import os
+
+# 将项目根目录加入 sys.path，以便直接运行该脚本
+sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
+
+from packages.core.logger import app_logger
+from packages.core.database import engine, Base, AsyncSessionLocal
+from packages.connectors.mock.us_mock import MockUSConnector
+
+async def init_db():
+    """初始化数据库表结构"""
+    app_logger.info("Initializing database tables...")
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    app_logger.info("Database tables initialized.")
+
+async def run_mock_job():
+    """运行 Mock Connector 验证链路"""
+    app_logger.info("Starting mock job...")
+    
+    async with AsyncSessionLocal() as session:
+        connector = MockUSConnector(session)
+        await connector.run()
+
+async def main():
+    try:
+        await init_db()
+        await run_mock_job()
+    except Exception as e:
+        app_logger.error(f"Execution failed: {e}")
+
+if __name__ == "__main__":
+    asyncio.run(main())

docker-compose.yml

@@ -0,0 +1,46 @@
+version: '3.8'
+
+services:
+  db:
+    image: postgres:15
+    container_name: customs_db
+    environment:
+      POSTGRES_USER: postgres
+      POSTGRES_PASSWORD: postgres
+      POSTGRES_DB: customs_data
+    ports:
+      - "5433:5432"
+    volumes:
+      - customs_pgdata:/var/lib/postgresql/data
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U postgres -d customs_data"]
+      interval: 5s
+      timeout: 5s
+      retries: 5
+    restart: always
+
+  api:
+    build: .
+    ports:
+      - "8000:8000"
+    depends_on:
+      db:
+        condition: service_healthy
+    volumes:
+      - .:/app
+    environment:
+      - DATABASE_URL=postgresql+asyncpg://postgres:postgres@db:5432/customs_data
+
+  scheduler:
+    build: .
+    depends_on:
+      db:
+        condition: service_healthy
+    volumes:
+      - .:/app
+    environment:
+      - DATABASE_URL=postgresql+asyncpg://postgres:postgres@db:5432/customs_data
+    command: ["bash", "-c", "python infrastructure/scheduler/main.py"]
+
+volumes:
+  customs_pgdata:

docs/国家与数据源接入清单.md

@@ -0,0 +1,159 @@
+# 国家与数据源接入清单
+
+## 1. 目标
+
+这份文档用于管理“国家覆盖是否全、更新是否及时、接入优先级如何排”的实际落地清单。
+
+规则：
+
+- 一个国家可对应多个数据源。
+- 必须同时记录“原始来源、合规状态、更新频率、字段完整度、接入难度”。
+- 未确认合法性的数据源，不进入正式商用。
+
+## 2. 数据源分级
+
+### A 类：优先
+
+- 官方 API
+- 官方开放数据下载
+- 官方统计接口
+
+特点：
+
+- 合规风险相对低
+- 稳定性相对高
+- 更适合长期运行
+
+### B 类：可用
+
+- 政府网页公开信息抓取
+- 半结构化文件下载
+- 公共统计平台
+
+特点：
+
+- 可做 MVP
+- 但页面变动风险更高
+
+### C 类：谨慎
+
+- 第三方聚合平台
+- 需账号、订阅或授权的数据源
+- 授权边界不清晰的数据源
+
+特点：
+
+- 可用于补齐覆盖
+- 但必须先过合规审查
+
+## 3. 国家优先级建议
+
+### P0：MVP 首批
+
+- [ ] 美国
+- [ ] 印度
+- [ ] 越南
+- [ ] 巴西
+- [ ] 墨西哥
+- [ ] 印度尼西亚
+- [ ] 土耳其
+- [ ] 菲律宾
+- [ ] 巴基斯坦
+- [ ] 欧盟公开统计源
+
+### P1：第二批
+
+- [ ] 马来西亚
+- [ ] 泰国
+- [ ] 阿联酋
+- [ ] 南非
+- [ ] 智利
+- [ ] 哥伦比亚
+- [ ] 阿根廷
+- [ ] 埃及
+
+### P2：后续补强
+
+- [ ] 日本
+- [ ] 韩国
+- [ ] 加拿大
+- [ ] 澳大利亚
+- [ ] 俄罗斯
+- [ ] 中东与非洲更多国家
+
+## 4. 首批国家（P0）初步数据源调研清单
+
+每接入一个国家，按下面模板补充，不要只记在代码里。
+
+### 4.1 美国 (United States)
+- 优先级：P0
+- 推荐数据源类型：CBP 官方 / Bill of Lading (提单数据) / PIERS 等第三方商业整合源
+- 难点与特点：美国官方海关（CBP）提供的是高度汇总数据，明细提单数据（B/L）需要走商业渠道或付费订阅。
+- 更新频率：日更（提单级别）/ 月更（宏观统计）
+- 字段完整度：高（包含真实企业名、重量、数量）
+- 合规状态：可用（提单数据属于公开记录，但需注意加州等隐私法对个人收件人的脱敏要求）
+- 适配器状态：未开始
+
+### 4.2 印度 (India)
+- 优先级：P0
+- 推荐数据源类型：Niryat Mitra / Zauba Trade (或同类第三方进出口平台)
+- 难点与特点：印度官方曾关闭过企业名级别的公开，部分第三方平台通过港口内部渠道获取，数据结构复杂，且常常存在拼写错误。
+- 更新频率：周更 / 月更
+- 字段完整度：中/高（HS 编码细致，企业名称需要强力清洗）
+- 合规状态：需排查第三方平台抓取风险
+- 适配器状态：未开始
+
+### 4.3 越南 (Vietnam)
+- 优先级：P0
+- 推荐数据源类型：越南海关总署 (General Department of Vietnam Customs) / GSO
+- 难点与特点：官方接口常变且有验证码，多数情况为汇总数据；明细需对接当地数据供应商。
+- 更新频率：月更
+- 字段完整度：中
+- 适配器状态：未开始
+
+### 4.4 巴西 (Brazil)
+- 优先级：P0
+- 推荐数据源类型：Siscomex / Comex Stat (官方外贸统计网)
+- 难点与特点：官方数据开放度极高，提供详细的 CSV/API 下载，但是企业信息（CNPJ）可能被哈希处理或部分隐藏。
+- 更新频率：月更
+- 字段完整度：高（金额、港口非常清晰）
+- 适配器状态：未开始
+
+### 4.5 墨西哥 (Mexico)
+- 优先级：P0
+- 推荐数据源类型：SAT (Servicio de Administración Tributaria) / SIAVI
+- 难点与特点：官方数据定期发布压缩包，解析难度中等，语言为西班牙语，需处理编码和字典翻译。
+- 适配器状态：未开始
+
+*(其他首批国家印尼、土耳其、菲律宾、巴基斯坦、欧盟在具体接入时按此模板补充)*
+
+## 5. 字段完整度标准
+
+### 高
+
+- 企业、商品、数量、金额、时间、国家、港口等核心字段大部分都有。
+
+### 中
+
+- 有主体交易信息，但部分字段缺失，仍可支持基础查询。
+
+### 低
+
+- 只能提供宏观统计或字段缺失严重，只能作为补充源。
+
+## 6. 更新优先级规则
+
+- [ ] 日更以上的数据源，优先做增量自动化。
+- [ ] 周更和月更的数据源，优先保证稳定同步，不追求过高轮询。
+- [ ] 公开源不稳定时，必须有备用源或人工补采预案。
+- [ ] 每个国家至少保留一个主源和一个备选源。
+
+## 7. 当前接入建议
+
+MVP 先做到以下状态：
+
+- [ ] 首批 10 个重点国家完成来源确认
+- [ ] 每个国家至少确定 1 个主源
+- [ ] 其中至少 5 个国家完成自动增量
+- [ ] 所有已接入国家都完成字段映射表
+- [ ] 所有已接入国家都完成合规标记

docs/海关数据项目-MVP方案.md

@@ -0,0 +1,525 @@
+# 海关数据项目 MVP 方案
+
+## 1. 项目目标
+
+### 1.1 目标
+
+- 做一个可持续运营的全球海关数据采集与查询平台。
+- 第一阶段先做 MVP，但底层架构按“后续可扩国家、可扩数据源、可扩更新频率”设计。
+- 核心要求：
+  - 数据覆盖尽可能全。
+  - 更新尽可能及时。
+  - 数据可追溯、可重跑、可纠错。
+  - 后续可支持商业查询、订阅预警、企业画像。
+
+### 1.2 MVP 定义
+
+MVP 不是一次把所有国家全部深度做完，而是先把平台骨架、统一数据模型、接入规范、调度体系、查询接口、质量控制体系搭起来，再优先接入高价值国家和公开可持续获取的数据源。
+
+MVP 的完成标准：
+
+- 有统一数据标准。
+- 有标准化采集流程。
+- 有定时更新机制。
+- 有去重、纠错、补采能力。
+- 有查询接口和基础后台。
+- 有国家接入扩展机制。
+
+## 2. 业务判断
+
+### 2.1 “数据要全”的现实定义
+
+全球海关数据不存在单一、完整、官方统一开放源。不同国家的数据开放程度、字段范围、更新频率、合规要求差异很大。
+
+所以“全”要拆成三层：
+
+- 国家覆盖全：尽可能覆盖更多国家。
+- 字段覆盖全：对每个国家尽可能保留原始字段，并映射到统一标准字段。
+- 时间覆盖全：尽可能做历史补采和持续增量更新。
+
+MVP 建议先实现：
+
+- 优先覆盖贸易数据价值高、公开度高、更新稳定的国家。
+- 每个国家保留“原始字段 + 标准字段 + 元数据”三套信息。
+- 所有采集任务必须支持历史回补与断点续跑。
+
+### 2.2 “及时更新”的现实定义
+
+不同国家海关数据不是实时流式开放，通常是日更、周更、月更，少数来源还会延迟发布。
+
+所以时效目标要按数据源分层：
+
+- T1：日更或准实时源，目标 1 小时到 6 小时内完成增量入库。
+- T2：周更源，目标当天完成同步。
+- T3：月更源，目标发布后 24 小时内完成同步。
+
+结论：产品层面强调“数据发布后最快同步”，不要承诺所有国家实时。
+
+## 3. 总体架构
+
+## 3.1 架构原则
+
+- 采集与清洗解耦。
+- 国家接入插件化。
+- 原始数据永远保留。
+- 标准模型统一，对外查询稳定。
+- 调度、重试、告警、审计独立。
+- 先保证稳定和可扩展，再逐步提高更新频率。
+
+### 3.2 推荐分层
+
+#### A. 数据源接入层
+
+负责从不同国家或第三方源采集原始数据：
+
+- 官方开放 API
+- 官方下载文件
+- 政府公开页面抓取
+- 第三方商业数据源
+- 手工导入源
+
+每个来源都实现统一接口：
+
+- `discover()`：发现新批次、新文件、新日期分片
+- `fetch()`：下载原始数据
+- `parse()`：解析成结构化记录
+- `checkpoint()`：记录进度
+
+#### B. 原始数据层
+
+保存未经破坏的原始结果：
+
+- 原始文件
+- 原始响应体
+- 原始记录 JSON
+- 来源元信息
+
+用途：
+
+- 出问题可回放
+- 后续字段扩展不用重新采集
+- 合规审计有依据
+
+#### C. 标准化处理层
+
+把不同国家的字段映射成统一结构：
+
+- 贸易时间
+- 进口/出口
+- 申报国
+- 起运国/目的国
+- HS 编码
+- 商品描述
+- 数量
+- 单位
+- 金额
+- 币种
+- 企业名称
+- 港口
+- 运输方式
+- 来源系统
+- 抓取批次
+
+这里要做：
+
+- 字段标准化
+- 币种处理
+- 时间格式统一
+- 国家/港口/单位字典映射
+- 企业名清洗
+- HS 编码规范化
+
+#### D. 数据质量层
+
+负责：
+
+- 去重
+- 缺失值识别
+- 字段异常检测
+- 同批次波动监控
+- 国家级采集成功率统计
+
+#### E. 服务层
+
+对内对外提供：
+
+- 查询 API
+- 后台任务管理
+- 数据源管理
+- 数据质量看板
+- 更新状态看板
+
+#### F. 调度与运维层
+
+负责：
+
+- 定时任务
+- 增量抓取
+- 历史补采
+- 失败重试
+- 死信任务
+- 告警通知
+
+## 4. 技术选型建议
+
+### 4.1 后端
+
+MVP 建议：
+
+- 采集任务：Python
+- API 服务：Python FastAPI
+- 调度：Prefect 或 Celery
+- 消息队列：Redis 或 RabbitMQ
+
+原因：
+
+- Python 生态适合抓取、解析、数据清洗。
+- FastAPI 适合快速出管理接口和查询接口。
+- Prefect 更适合数据工作流管理；如果更重视成熟队列消费，也可以用 Celery。
+
+### 4.2 数据存储
+
+- 结构化主库：PostgreSQL
+- 检索增强：OpenSearch 或 Elasticsearch
+- 原始文件存储：对象存储
+- 缓存：Redis
+
+建议：
+
+- PostgreSQL 存标准记录、任务、元数据、字典表。
+- OpenSearch 做商品描述、企业名、模糊搜索。
+- 对象存储保留原始文件和原始响应。
+
+### 4.3 前端
+
+MVP 前端建议拆两部分：
+
+- 管理后台：任务监控、国家接入管理、质量看板
+- 查询前台：企业、商品、国家、时间维度查询
+
+技术可选：
+
+- Web：Next.js 或 React
+- 组件体系：统一一套后台组件和表格筛选能力
+
+## 5. 数据模型设计
+
+### 5.1 原始记录表
+
+建议保留：
+
+- 来源国家
+- 来源系统
+- 采集时间
+- 原始主键
+- 原始 JSON
+- 文件地址
+- 批次号
+- 哈希指纹
+- 解析状态
+
+### 5.2 标准贸易记录表
+
+核心字段建议：
+
+- `record_id`
+- `source_country`
+- `trade_direction`
+- `trade_date`
+- `importer_name`
+- `exporter_name`
+- `shipper_name`
+- `consignee_name`
+- `origin_country`
+- `destination_country`
+- `departure_port`
+- `arrival_port`
+- `hs_code`
+- `product_name`
+- `quantity`
+- `quantity_unit`
+- `amount`
+- `currency`
+- `weight`
+- `weight_unit`
+- `transport_mode`
+- `data_source`
+- `source_record_id`
+- `first_seen_at`
+- `last_seen_at`
+- `batch_no`
+
+### 5.3 去重策略
+
+不能只按单一字段去重，建议做组合指纹：
+
+- 国家
+- 日期
+- 企业名
+- HS 编码
+- 商品描述
+- 数量
+- 金额
+- 港口
+
+对无法完全确定重复的数据，保留多版本并打疑似重复标记，不直接硬删。
+
+## 6. 国家接入策略
+
+### 6.1 接入优先级
+
+MVP 不建议一上来做全球同时深挖，建议按三层推进：
+
+- P0：高价值、高公开度、高稳定性国家
+- P1：有公开源但解析复杂的国家
+- P2：公开性差、需商业采购或特殊合规处理的国家
+
+### 6.2 推荐第一批国家
+
+优先考虑：
+
+- 美国
+- 印度
+- 越南
+- 巴西
+- 墨西哥
+- 印度尼西亚
+- 土耳其
+- 菲律宾
+- 巴基斯坦
+- 部分欧盟国家公开贸易统计源
+
+选择标准：
+
+- 对外贸业务价值高
+- 数据有公开入口或稳定来源
+- 更新频率可接受
+- 字段较完整
+
+### 6.3 国家适配器机制
+
+每个国家一个独立适配器目录，包含：
+
+- 源配置
+- 抓取逻辑
+- 解析逻辑
+- 字段映射
+- 增量规则
+- 限流规则
+- 异常处理
+
+这样后续新增国家不会破坏主流程。
+
+## 7. 更新机制设计
+
+### 7.1 三类任务
+
+- 历史回补任务
+- 日常增量任务
+- 修复重跑任务
+
+### 7.2 更新策略
+
+每个国家配置独立更新策略：
+
+- 拉取周期
+- 可补采窗口
+- 单次抓取日期跨度
+- 并发数
+- 限速
+- 失败重试次数
+- 是否需要人工校验
+
+### 7.3 及时更新的关键点
+
+- 用“源发现器”先判断是否有新批次，避免盲目全量拉取。
+- 对高频国家使用小时级轮询。
+- 对低频国家用日级或周级调度。
+- 增量失败时只重跑失败分片，不重跑整国全量。
+- 每次入库都写入批次号和时间戳，便于校验延迟。
+
+## 8. 合规与风控
+
+必须单独设计，不要等上线前再补。
+
+重点包括：
+
+- 各国家数据源的使用条款
+- robots 与访问频率限制
+- 是否允许商业使用
+- 是否包含个人敏感信息
+- 是否需要脱敏展示
+- 是否需要地区合规隔离
+
+原则：
+
+- 原始数据保留，但展示和对外提供要按合规规则做权限控制。
+- 对来源不明确或授权不清晰的数据源，先标红，不进入正式商用流程。
+
+## 9. MVP 范围建议
+
+### 9.1 必做
+
+- 国家适配器框架
+- 10 个左右重点国家接入
+- 原始数据存储
+- 标准字段映射
+- 去重机制
+- 定时增量更新
+- 查询 API
+- 后台任务管理
+- 更新状态看板
+- 失败告警
+
+### 9.2 暂缓
+
+- 全量全球深度覆盖
+- 复杂画像算法
+- 智能推荐
+- 高级 BI 分析
+- 客户级权限体系
+- 大规模多租户计费
+
+## 10. 实施阶段建议
+
+### 阶段 1：底座
+
+- 统一数据模型
+- 任务框架
+- 原始数据存储
+- 调度中心
+- 查询 API 骨架
+
+### 阶段 2：首批国家
+
+- 接入 5 到 10 个高价值国家
+- 打通历史补采与日常增量
+- 建立质量看板
+
+### 阶段 3：产品可用
+
+- 搜索查询
+- 企业维度筛选
+- 商品维度筛选
+- 时间维度筛选
+- 导出能力
+
+### 阶段 4：扩国家
+
+- 按模板快速接入更多国家
+- 按国家差异优化时效和质量
+
+## 11. 推荐目录结构
+
+```text
+project/
+  apps/
+    api/
+    admin/
+    worker/
+  packages/
+    core/
+    connectors/
+    normalizers/
+    dictionaries/
+    shared/
+  infrastructure/
+    scheduler/
+    monitoring/
+  docs/
+```
+
+说明：
+
+- `api`：查询接口与管理接口
+- `admin`：后台
+- `worker`：采集、清洗、入库任务
+- `connectors`：国家适配器
+- `normalizers`：标准化逻辑
+- `dictionaries`：国家、港口、币种、单位、HS 映射
+
+## 12. 最终建议
+
+这个项目的核心，不是“先把页面做出来”，而是先把以下 4 个底座做稳：
+
+- 国家适配器标准
+- 原始数据留存
+- 标准模型
+- 调度与质量控制
+
+如果这 4 个底座先做对，后面扩国家、提时效、做商业化会比较顺。
+
+如果这 4 个底座先做错，后面国家越多，系统越难维护。
+
+## 13. 架构设计补充建议（进阶增强）
+
+作为架构视角的补充，为了保证系统在长期海量数据采集和存储下的稳定性与成本可控，建议在 MVP 阶段就将以下底层机制纳入设计（不一定立刻完整实现，但接口要留好）：
+
+### 13.1 动态代理池与反爬隔离 (Anti-Crawler & Proxy Pool)
+- **痛点**：全球各国的海关官网或第三方源都有严格的访问频控和反爬机制（如 Cloudflare、滑块验证码）。
+- **建议**：在 `connectors` 层下方封装统一的 `Request Client`。
+  - 必须剥离业务代码与网络请求代码。
+  - 接口原生支持动态切换代理池（住宅 IP、数据中心 IP）。
+  - 预留 JS 渲染（Playwright/Puppeteer）和打码平台对接能力，应对突发的反爬升级。
+
+### 13.2 解析器版本控制与数据重放 (Schema Evolution)
+- **痛点**：海关数据源的网页结构或 API 字段经常会毫无预警地变更，导致旧的解析器报错或解析出脏数据。
+- **建议**：
+  - 每个国家的数据 `parser` 必须维护版本号（如 `v1.0.1`），并在落库的记录中带上 `parser_version`。
+  - 一旦发现解析错误或逻辑更新，由于原始数据（HTML/JSON）仍存放在对象存储中，只需升级 `parser` 版本，利用 `worker` 对指定的历史批次触发“重放解析”即可，无需重新请求对方接口。
+
+### 13.3 数据血缘追踪 (Data Lineage)
+- **痛点**：客户或内部运营发现某条标准数据的金额或企业名异常时，需要快速排查是源头本身错了，还是我们解析、汇率换算时出错了。
+- **建议**：
+  - 标准表中的 `record_id` 必须能直接关联到 `source_record_id`。
+  - 通过 `batch_no` 直接在对象存储中定位到那一次抓取的原始响应切片，做到“所见即所得”的防甩锅和可追溯。
+
+### 13.4 存储降本与冷热分离架构
+- **痛点**：海关数据是典型的时间序列数据，越老的数据查询频率越低，但全量累积放在关系型数据库中成本极高且拖慢性能。
+- **建议**：
+  - **热数据**（近 1-2 年）：存 PG + OpenSearch，支持高性能多维检索。
+  - **温数据**（2-5 年）：存 PG（按时间分区表），OpenSearch 剔除部分非核心索引或降低副本数。
+  - **冷数据**（5 年前）：归档导出为 Parquet 格式存入 S3（对象存储），并通过类似 Athena/DuckDB 等机制按需查询，释放高昂的数据库内存。
+
+## 14. MVP 阶段开发计划预估
+
+为了保证项目可落地，建议将 MVP 切分为 4 个冲刺（Sprint），每个 Sprint 为期 1-2 周（具体根据投入人力调整）。核心原则是**“底层先行，串联验证，最后扩容”**。
+
+### Sprint 1: 核心底座与架构初始化
+**目标**：搭建工程骨架，跑通从“模拟抓取 -> 原始存储 -> 解析标准化 -> 关系型入库”的单链路。
+- **任务 1**：环境搭建（Python + FastAPI + PostgreSQL + Redis 基础容器化）。
+- **任务 2**：核心包（`packages/core`）开发，包括基础数据模型（ORM 定义）、统一日志、配置管理。
+- **任务 3**：实现基础的 `Request Client`（支持代理接入机制）。
+- **任务 4**：设计并实现第一个国家（如美国或巴西）的 Mock `Connector`，跑通下载和存储逻辑（打通 S3 或本地 Mock 存储）。
+- **交付物**：可运行的工程脚手架、数据库表结构初始化脚本、单条链路的日志与存储验证。
+
+### Sprint 2: 真实数据源接入与解析引擎
+**目标**：接入 3 个真实高价值国家，完成核心字段标准化与清洗逻辑。
+- **任务 1**：开发并调试 3 个首批国家（如美国、巴西、越南）的真实 `Connector`。
+- **任务 2**：开发标准化组件（`packages/normalizers`），实现企业名初步清洗、日期格式化。
+- **任务 3**：建立基础映射字典（`packages/dictionaries`），如国家代码（ISO 3166）、货币（ISO 4217）、单位转换。
+- **任务 4**：实现血缘追踪机制（`source_record_id` 与 `batch_no` 绑定落库）。
+- **交付物**：3 个国家的增量数据能稳定入库，并在 PG 中形成标准化的贸易记录。
+
+### Sprint 3: 任务调度与质量监控
+**目标**：脱离手动脚本，实现自动化、定时调度与异常告警。
+- **任务 1**：引入调度器（Prefect / Celery），将 3 个国家的抓取与解析封装为定时任务。
+- **任务 2**：实现历史回补与断点续跑逻辑（`checkpoint` 机制）。
+- **任务 3**：开发数据质量监控模块（如缺失率统计、单次批次数量波动告警）。
+- **任务 4**：对接飞书/钉钉/企业微信或邮件的异常通知。
+- **交付物**：无人值守的自动化采集链路，具备失败重试与告警能力。
+
+### Sprint 4: 服务层 API 与扩容准备
+**目标**：提供对外的查询能力，并将国家扩容至首批 10 个。
+- **任务 1**：开发 `apps/api` 查询接口，支持按企业、国家、HS 编码、时间段的多维查询。
+- **任务 2**：接入剩余 7 个 P0 国家（印度、墨西哥等），验证并优化 `Connector` ���复用度。
+- **任务 3**：引入 OpenSearch（如果 PG 的查询性能达到瓶颈），实现商品描述和企业名的模糊检索。
+- **任务 4**：提供简单的后台看板接口（`apps/admin`），展示各国家最近一次更新时间与成功率。
+- **交付物**：首批 10 国数据全自动化运转，提供标准查询 API，MVP 闭环完成。
+
+## 历史数据回溯策略 (Backfill)
+
+为了应对海关数据的时间跨度问题，系统内置了专门的历史数据回溯机制：
+- **执行方式**：通过独立的回溯脚本（`apps/worker/run_backfill.py`）执行，不与日常增量定时任务混合，避免影响最新数据的获取时效。
+- **拆分策略**：脚本按自然月切分时间窗口。如果用户要求回溯过去 5 年的数据，系统会自动将其切分为 60 个月的批次，分批请求海关源或公开数据集，防止单次请求超时或触发反爬。
+- **数据对齐**：回溯抓取的数据与日常增量一样，必须经过标准化的 Pipeline（`save_raw` -> `parse` -> `normalize` -> `save_standard`）。这意味着 10 年前的旧公司名称也会被最新的企业名清洗字典（`normalizers/company.py`）清洗为现代标准名。

infrastructure/monitoring/alert.py

@@ -0,0 +1,16 @@
+from packages.core.logger import app_logger
+
+def send_alert(title: str, message: str, level: str = "warning"):
+    """
+    通用告警模块：
+    未来可在此处接入钉钉/飞书/企业微信机器人的 Webhook。
+    当前 MVP 阶段直接输出到高优先级日志。
+    """
+    alert_msg = f"🚨 [ALERT - {level.upper()}] {title}: {message}"
+    
+    if level == "critical" or level == "error":
+        app_logger.error(alert_msg)
+    else:
+        app_logger.warning(alert_msg)
+        
+    # TODO: 实现 HTTP POST 调用飞书/钉钉 Webhook

infrastructure/monitoring/quality.py

@@ -0,0 +1,43 @@
+from sqlalchemy import select, func
+from packages.core.database import AsyncSessionLocal
+from packages.core.models import StandardTradeRecord
+from packages.core.logger import app_logger
+from infrastructure.monitoring.alert import send_alert
+
+async def check_batch_quality(batch_no: str):
+    """
+    数据质量巡检：
+    在每批次数据采集并标准化入库后执行，检查是否有大规模字段缺失或总量异常。
+    """
+    app_logger.info(f"Starting quality check for batch: {batch_no}")
+    
+    async with AsyncSessionLocal() as session:
+        # 1. 统计批次总记录数
+        total_stmt = select(func.count()).select_from(StandardTradeRecord).where(
+            StandardTradeRecord.batch_no == batch_no
+        )
+        total_result = await session.execute(total_stmt)
+        total = total_result.scalar() or 0
+
+        if total == 0:
+            send_alert("数据空跑告警", f"批次 {batch_no} 抓取入库的标准化记录数为 0！", level="error")
+            return
+
+        # 2. 统计 HS 编码缺失数
+        missing_hs_stmt = select(func.count()).select_from(StandardTradeRecord).where(
+            StandardTradeRecord.batch_no == batch_no,
+            (StandardTradeRecord.hs_code == None) | (StandardTradeRecord.hs_code == "")
+        )
+        missing_result = await session.execute(missing_hs_stmt)
+        missing_hs = missing_result.scalar() or 0
+
+        missing_rate = missing_hs / total
+        app_logger.info(f"Batch {batch_no} quality metrics - Total: {total}, Missing HS: {missing_hs} ({missing_rate:.2%})")
+
+        # 3. 触发阈值告警 (例如缺失率超过 30%)
+        if missing_rate > 0.3:
+            send_alert(
+                "字段大面积缺失告警", 
+                f"批次 {batch_no} 的 HS 编码缺失率高达 {missing_rate:.2%} (阈值 30%)，可能解析器失效或源端改版！",
+                level="warning"
+            )

infrastructure/scheduler/main.py

@@ -0,0 +1,59 @@
+import asyncio
+import sys
+import os
+
+# 将项目根目录加入 sys.path
+sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
+
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+from packages.core.logger import app_logger
+
+# 引入已有的采集任务
+from apps.worker.run_mock import run_mock_job
+from apps.worker.run_brazil import run_brazil_job
+from apps.worker.run_extended import run_extended_mock_jobs
+
+async def scheduled_us_job():
+    try:
+        app_logger.info("Triggering US Mock Sync Job via Scheduler...")
+        await run_mock_job()
+    except Exception as e:
+        app_logger.error(f"Scheduled US job failed: {e}")
+
+async def scheduled_br_job():
+    try:
+        app_logger.info("Triggering Brazil Sync Job via Scheduler...")
+        await run_brazil_job()
+    except Exception as e:
+        app_logger.error(f"Scheduled BR job failed: {e}")
+
+async def scheduled_extended_job():
+    try:
+        app_logger.info("Triggering Extended Countries Sync Job via Scheduler...")
+        await run_extended_mock_jobs()
+    except Exception as e:
+        app_logger.error(f"Scheduled Extended job failed: {e}")
+
+def start_scheduler():
+    """
+    启动任务调度中心
+    """
+    scheduler = AsyncIOScheduler()
+    
+    # MVP 阶段用于演示，设置较短的定时：
+    scheduler.add_job(scheduled_us_job, 'interval', minutes=2, id="sync_us_data")
+    scheduler.add_job(scheduled_br_job, 'interval', minutes=3, id="sync_br_data")
+    # 新增的 7 个国家，每 4 分钟跑一次批处理
+    scheduler.add_job(scheduled_extended_job, 'interval', minutes=4, id="sync_extended_data")
+
+    scheduler.start()
+    app_logger.info("Scheduler started successfully. Waiting for jobs to execute...")
+
+    # 保持主线程事件循环不退出
+    try:
+        asyncio.get_event_loop().run_forever()
+    except (KeyboardInterrupt, SystemExit):
+        app_logger.info("Scheduler shutting down...")
+
+if __name__ == "__main__":
+    start_scheduler()

init_db.py

@@ -0,0 +1,22 @@
+import asyncio
+from sqlalchemy.ext.asyncio import create_async_engine
+from packages.core.config import settings
+from packages.core.models import Base
+
+async def init_models():
+    # 使用 PostgreSQL DSN
+    engine = create_async_engine(settings.DATABASE_URL, echo=True)
+    
+    # 因为有 PostGIS 扩展依赖，最好确保扩展已被创建
+    # 正常 postgis image 默认在 public 下已有 postgis
+    
+    async with engine.begin() as conn:
+        # 删除所有旧表（如果是全新项目，这里清理掉之前海关项目的旧表）
+        await conn.run_sync(Base.metadata.drop_all)
+        # 创建所有新表
+        await conn.run_sync(Base.metadata.create_all)
+        
+    print("数据库表结构初始化成功！")
+    
+if __name__ == "__main__":
+    asyncio.run(init_models())

packages/connectors/base.py

@@ -0,0 +1,162 @@
+import hashlib
+import json
+import uuid
+from datetime import datetime, timezone
+from typing import List, Dict, Any
+from sqlalchemy.ext.asyncio import AsyncSession
+from packages.core.logger import app_logger
+from packages.core.models import RawTradeRecord, StandardTradeRecord
+
+class BaseConnector:
+    """
+    国家数据源适配器基类，规定了标准执行流程：
+    discover -> fetch -> parse -> save_raw -> normalize -> save_standard
+    """
+    
+    country_code: str = "UNKNOWN"
+    source_system: str = "UNKNOWN"
+    parser_version: str = "1.0.0"
+
+    def __init__(self, session: AsyncSession):
+        self.session = session
+        self.batch_no = f"{self.country_code}_{datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')}"
+
+    async def discover(self) -> List[Any]:
+        """发现新数据批次或需要抓取的任务切片"""
+        raise NotImplementedError
+
+    async def fetch(self, task_slice: Any) -> Dict[str, Any]:
+        """执行抓取并返回原始数据"""
+        raise NotImplementedError
+
+    async def parse(self, raw_data: Any) -> List[Dict[str, Any]]:
+        """
+        [子类必须实现] 
+        解析并清洗原始数据响应，提取出需要留存为 RawTradeRecord 的记录列表
+        注意：此时不关联数据库模型，仅返回字典列表
+        """
+        raise NotImplementedError
+
+    async def save_raw(self, raw_items: List[Dict[str, Any]]) -> List[RawTradeRecord]:
+        """将原始数据留存入库，返回 RawTradeRecord 实例列表用于血缘追踪"""
+        raw_records = []
+        for item in raw_items:
+            raw_json_str = json.dumps(item, ensure_ascii=False, sort_keys=True)
+            hash_fp = hashlib.sha256(raw_json_str.encode('utf-8')).hexdigest()
+            
+            record = RawTradeRecord(
+                id=str(uuid.uuid4()),
+                batch_no=self.batch_no,
+                source_country=self.country_code,
+                source_system=self.source_system,
+                raw_json=item,
+                content_hash=hash_fp
+            )
+            raw_records.append(record)
+        
+        if raw_records:
+            # 为防止重跑导致内容指纹重复报错，MVP 阶段忽略冲突
+            from sqlalchemy.dialects.postgresql import insert
+            
+            # 首先根据 content_hash 查询已经存在的记录
+            hashes = [r.content_hash for r in raw_records]
+            from sqlalchemy import select
+            stmt = select(RawTradeRecord).where(RawTradeRecord.content_hash.in_(hashes))
+            result = await self.session.execute(stmt)
+            existing_records = {r.content_hash: r.id for r in result.scalars()}
+            
+            # 更新已存在记录的 ID，对于不存在的插入
+            to_insert = []
+            for r in raw_records:
+                if r.content_hash in existing_records:
+                    r.id = existing_records[r.content_hash]
+                else:
+                    to_insert.append({
+                        "id": r.id,
+                        "batch_no": r.batch_no,
+                        "source_country": r.source_country,
+                        "source_system": r.source_system,
+                        "raw_json": r.raw_json,
+                        "content_hash": r.content_hash
+                    })
+                    
+            if to_insert:
+                stmt = insert(RawTradeRecord).values(to_insert).on_conflict_do_nothing(index_elements=["content_hash"])
+                await self.session.execute(stmt)
+                await self.session.flush()
+            
+        return raw_records
+
+    async def normalize(self, raw: RawTradeRecord) -> StandardTradeRecord:
+        """
+        [子类必须实现]
+        将单条原始数据记录（含 DB ID）转换为标准数据记录模型
+        """
+        raise NotImplementedError
+
+    async def save_standard(self, std_records: List[StandardTradeRecord]):
+        """保存标准数据"""
+        if std_records:
+            from sqlalchemy.dialects.postgresql import insert
+            stmt = insert(StandardTradeRecord).values([{
+                "record_id": r.record_id,
+                "source_record_id": r.source_record_id,
+                "batch_no": r.batch_no,
+                "source_country": r.source_country,
+                "trade_direction": r.trade_direction,
+                "trade_date": r.trade_date,
+                "importer_name": r.importer_name,
+                "exporter_name": r.exporter_name,
+                "hs_code": r.hs_code,
+                "product_name": r.product_name,
+                "amount": r.amount,
+                "currency": r.currency,
+                "weight": r.weight,
+                "weight_unit": r.weight_unit,
+                "origin_country": r.origin_country,
+                "destination_country": r.destination_country,
+                "departure_port": r.departure_port,
+                "arrival_port": r.arrival_port,
+                "transport_mode": r.transport_mode
+            } for r in std_records]).on_conflict_do_nothing(index_elements=["record_id"])
+            
+            await self.session.execute(stmt)
+            await self.session.commit()
+
+    async def run(self):
+        """执行完整链路"""
+        app_logger.info(f"[{self.country_code}] Starting job, batch_no: {self.batch_no}")
+        
+        try:
+            task_slices = await self.discover()
+            for ts in task_slices:
+                raw_data = await self.fetch(ts)
+                parsed_items = await self.parse(raw_data)
+                
+                if not isinstance(parsed_items, list):
+                    parsed_items = [parsed_items]
+                
+                # 1. 留存原始数据
+                raw_records = await self.save_raw(parsed_items)
+                app_logger.info(f"[{self.country_code}] Saved {len(raw_records)} raw records")
+                
+                # 2. 数据标准化与血缘绑定
+                std_records = []
+                for raw in raw_records:
+                    std_rec = await self.normalize(raw)
+                    std_records.append(std_rec)
+                
+                # 3. 保存标准数据
+                await self.save_standard(std_records)
+                app_logger.info(f"[{self.country_code}] Saved {len(std_records)} standard records")
+                
+                # 4. 触发数据质量巡检
+                from infrastructure.monitoring.quality import check_batch_quality
+                await check_batch_quality(self.batch_no)
+                
+            app_logger.info(f"[{self.country_code}] Job completed successfully")
+            
+        except Exception as e:
+            await self.session.rollback()
+            app_logger.error(f"[{self.country_code}] Job failed: {str(e)}")
+            raise e

packages/connectors/brazil.py

@@ -0,0 +1,131 @@
+import uuid
+from typing import List, Dict, Any
+from datetime import datetime
+from packages.connectors.base import BaseConnector
+from packages.core.models import RawTradeRecord, StandardTradeRecord
+from packages.dictionaries.brazil import CURRENCY_MAP, UNIT_MAP, TRANSPORT_MODE_MAP
+from packages.normalizers.company import clean_company_name
+
+class BrazilComexStatConnector(BaseConnector):
+    """
+    巴西 Comex Stat 官方统计数据接入适配器
+    这是一个基于 API / CSV 的真实数据源结构模拟
+    """
+    country_code = "BR"
+    source_system = "COMEX_STAT_API"
+    parser_version = "v1.1.0"
+
+    async def discover(self) -> List[Any]:
+        # 模拟发现 2024 年 5 月的数据批次
+        return [{"year": 2024, "month": 5}]
+
+    async def fetch(self, task_slice: Any) -> Dict[str, Any]:
+        # 模拟调用巴西统计局接口返回的原始数据结构（葡萄牙语字段）
+        # 真实情况这里会使用 self.http_client 进行 requests
+        import random
+        from datetime import datetime, timedelta
+        
+        random_days_ago_1 = random.randint(0, 365 * 10)
+        simulated_date_1 = (datetime.now() - timedelta(days=random_days_ago_1)).strftime("%Y-%m-%d")
+        
+        random_days_ago_2 = random.randint(0, 365 * 10)
+        simulated_date_2 = (datetime.now() - timedelta(days=random_days_ago_2)).strftime("%Y-%m-%d")
+        
+        return {
+            "metadata": {
+                "status": "success",
+                "period": f"{task_slice['year']}-{task_slice['month']:02d}"
+            },
+            "data": [
+                {
+                    "id_registro": "BR-202405-001",
+                    "data_registro": simulated_date_1,
+                    "tipo_operacao": "IMPORTACAO", # 进口
+                    "cnpj_importador": "00.000.000/0001-91",
+                    "nome_importador": "BANCO DO BRASIL S.A.",
+                    "nome_exportador": "IBM CORPORATION",
+                    "ncm_codigo": "8471.30.12", # HS 编码在巴西叫 NCM
+                    "descricao_mercadoria": "MAQUINAS AUTOMATICAS PARA PROCESSAMENTO DE DADOS",
+                    "valor_fob": "1500000.00",
+                    "moeda": "DÓLAR",
+                    "peso_liquido": "2000",
+                    "unidade_medida": "QUILOGRAMA",
+                    "pais_origem": "US",
+                    "porto_desembarque": "SANTOS",
+                    "via_transporte": "MARÍTIMO"
+                },
+                {
+                    "id_registro": "BR-202405-002",
+                    "data_registro": simulated_date_2,
+                    "tipo_operacao": "EXPORTACAO", # 出口
+                    "cnpj_exportador": "11.111.111/0001-11",
+                    "nome_exportador": "VALE S.A.",
+                    "nome_importador": "BAOSTEEL GROUP",
+                    "ncm_codigo": "2601.11.00",
+                    "descricao_mercadoria": "MINERIOS DE FERRO E SEUS CONCENTRADOS",
+                    "valor_fob": "8500000.00",
+                    "moeda": "DÓLAR",
+                    "peso_liquido": "150000",
+                    "unidade_medida": "TONELADA",
+                    "pais_destino": "CN",
+                    "porto_embarque": "VITORIA",
+                    "via_transporte": "MARÍTIMO"
+                }
+            ]
+        }
+
+    async def parse(self, raw_data: Any) -> List[Dict[str, Any]]:
+        # 实际的巴西接口返回的是一个字典 {"data": [...]}，或者列表
+        if isinstance(raw_data, dict):
+            return raw_data.get("data", [])
+        elif isinstance(raw_data, list):
+            return raw_data
+        else:
+            return []
+
+    async def normalize(self, raw: RawTradeRecord) -> StandardTradeRecord:
+        data = raw.raw_json
+        
+        # 判断贸易方向
+        op_type = data.get("tipo_operacao", "").upper()
+        trade_direction = "import" if op_type == "IMPORTACAO" else "export"
+        
+        # 解析企业名并清洗
+        if trade_direction == "import":
+            importer = data.get("nome_importador", "")
+            exporter = data.get("nome_exportador", "")
+        else:
+            importer = data.get("nome_importador", "")
+            exporter = data.get("nome_exportador", "")
+            
+        importer_clean = clean_company_name(importer)
+        exporter_clean = clean_company_name(exporter)
+        
+        # 字典映射
+        currency = CURRENCY_MAP.get(data.get("moeda", "").upper(), "UNKNOWN")
+        weight_unit = UNIT_MAP.get(data.get("unidade_medida", "").upper(), "UNKNOWN")
+        transport_mode = TRANSPORT_MODE_MAP.get(data.get("via_transporte", "").upper(), "UNKNOWN")
+        
+        record_id = str(uuid.uuid4())
+        
+        return StandardTradeRecord(
+            record_id=record_id,
+            source_record_id=raw.id,
+            batch_no=raw.batch_no,
+            source_country=self.country_code,
+            trade_direction=trade_direction,
+            trade_date=datetime.strptime(data.get("data_registro"), "%Y-%m-%d"),
+            importer_name=importer_clean,
+            exporter_name=exporter_clean,
+            hs_code=data.get("ncm_codigo", "").replace(".", ""), # 巴西 NCM 带点，标准化去掉
+            product_name=data.get("descricao_mercadoria"),
+            amount=float(data.get("valor_fob", 0)),
+            currency=currency,
+            weight=float(data.get("peso_liquido", 0)),
+            weight_unit=weight_unit,
+            origin_country=data.get("pais_origem") if trade_direction == "import" else self.country_code,
+            destination_country=data.get("pais_destino") if trade_direction == "export" else self.country_code,
+            departure_port=data.get("porto_embarque"),
+            arrival_port=data.get("porto_desembarque"),
+            transport_mode=transport_mode
+        )

packages/connectors/mock/extended_mock.py

@@ -0,0 +1,87 @@
+import uuid
+import random
+from datetime import datetime, timedelta
+from packages.connectors.base import BaseConnector
+
+# 新国家的模拟配置字典
+COUNTRY_CONFIGS = {
+    "ID": {"currency": "IDR", "port": "JAKARTA", "importers": ["PT. ASTRA", "PT. INDOFOOD", "PT. TELEKOMUNIKASI"]},
+    "TH": {"currency": "THB", "port": "BANGKOK", "importers": ["CP GROUP", "PTT PCL", "SIAM CEMENT"]},
+    "VN": {"currency": "VND", "port": "HO CHI MINH", "importers": ["VINAMILK", "VIETTEL", "PETROVIETNAM"]},
+    "MY": {"currency": "MYR", "port": "PORT KLANG", "importers": ["PETRONAS", "MAYBANK", "TENAGA NASIONAL"]},
+    "NZ": {"currency": "NZD", "port": "AUCKLAND", "importers": ["FONTERRA", "AIR NEW ZEALAND", "FLETCHER BUILDING"]},
+    "AE": {"currency": "AED", "port": "DUBAI", "importers": ["EMIRATES", "DP WORLD", "EMAAR PROPERTIES"]},
+    "DE": {"currency": "EUR", "port": "HAMBURG", "importers": ["VOLKSWAGEN", "SIEMENS", "BOSCH"]}
+}
+
+from packages.core.models import RawTradeRecord, StandardTradeRecord
+
+class ExtendedMockConnector(BaseConnector):
+    """
+    通用模拟数据适配器，支持根据传入的国家代码动态生成该国的模拟数据
+    """
+    def __init__(self, country_code: str, session):
+        super().__init__(session=session)
+        self.country_code = country_code
+        self.source_system = f"Mock_{country_code}_Customs"
+        self.config = COUNTRY_CONFIGS[country_code]
+
+    async def discover(self) -> list:
+        # 每次模拟发现 2-5 条新数据
+        count = random.randint(2, 5)
+        return [{"id": str(uuid.uuid4())} for _ in range(count)]
+
+    async def parse(self, raw_data: dict) -> dict:
+        return raw_data
+        
+    async def fetch(self, tasks: list) -> list:
+        results = []
+        for t in tasks:
+            # tasks 里如果是一维数组的字典
+            item_id = t.get("id") if isinstance(t, dict) else t
+            
+            # 模拟生成过去 10 年内随机某一天的数据
+            random_days_ago = random.randint(0, 365 * 10)
+            simulated_date = (datetime.now() - timedelta(days=random_days_ago)).strftime("%Y-%m-%d")
+            
+            results.append({
+                "id": item_id,
+                "trade_type": random.choice(["import", "export"]),
+                "date": simulated_date,
+                "importer": random.choice(self.config["importers"]),
+                "exporter": "GLOBAL TRADING LLC",
+                "hs": f"8471{random.randint(1000, 9999)}",
+                "desc": "COMMERCIAL GOODS / MACHINERY",
+                "value": random.randint(10000, 500000),
+                "curr": self.config["currency"],
+                "qty": random.randint(1000, 5000),
+                "unit": "KG",
+                "port": self.config["port"]
+            })
+        return results
+
+    async def normalize(self, raw: RawTradeRecord) -> StandardTradeRecord:
+        data = raw.raw_json
+        is_import = data["trade_type"] == "import"
+        
+        return StandardTradeRecord(
+            record_id=str(uuid.uuid4()),
+            source_record_id=raw.id,
+            batch_no=raw.batch_no,
+            source_country=self.country_code,
+            trade_direction=data["trade_type"],
+            trade_date=datetime.strptime(data["date"], "%Y-%m-%d"),
+            importer_name=data["importer"],
+            exporter_name=data["exporter"],
+            hs_code=data["hs"],
+            product_name=data["desc"],
+            amount=float(data["value"]),
+            currency=data["curr"],
+            weight=float(data["qty"]),
+            weight_unit=data["unit"],
+            origin_country="CN" if is_import else self.country_code,
+            destination_country=self.country_code if is_import else "US",
+            departure_port="SHANGHAI" if is_import else data["port"],
+            arrival_port=data["port"] if is_import else "LOS ANGELES",
+            transport_mode="SEA"
+        )

packages/connectors/mock/us_mock.py

@@ -0,0 +1,87 @@
+from typing import List, Dict, Any
+import uuid
+from datetime import datetime
+from packages.connectors.base import BaseConnector
+from packages.core.models import RawTradeRecord, StandardTradeRecord
+
+class MockUSConnector(BaseConnector):
+    """
+    模拟美国海关提单数据接入
+    用于跑通 MVP 底层链路，不发起真实网络请求
+    """
+    country_code = "US"
+    source_system = "US_CBP_MOCK"
+    parser_version = "v1.0.0"
+
+    async def discover(self) -> List[Any]:
+        # 模拟发现了一个日期的切片任务
+        return ["2024-05-28"]
+
+    async def fetch(self, task_slice: Any) -> Dict[str, Any]:
+        import random
+        from datetime import timedelta
+        # 模拟抓取过程，返回包含 10 年内随机时间的原始 JSON 数据
+        random_days_ago_1 = random.randint(0, 365 * 10)
+        simulated_date_1 = (datetime.now() - timedelta(days=random_days_ago_1)).strftime("%Y-%m-%d")
+        
+        random_days_ago_2 = random.randint(0, 365 * 10)
+        simulated_date_2 = (datetime.now() - timedelta(days=random_days_ago_2)).strftime("%Y-%m-%d")
+
+        return {
+            "status": "ok",
+            "date": task_slice,
+            "results": [
+                {
+                    "id": "B/L-1001",
+                    "consignee": "APPLE INC.",
+                    "shipper": "FOXCONN TECH",
+                    "description": "SMARTPHONES",
+                    "qty": "1000",
+                    "qty_unit": "PCS",
+                    "port_of_lading": "SHENZHEN",
+                    "port_of_unlading": "LOS ANGELES",
+                    "date": simulated_date_1
+                },
+                {
+                    "id": "B/L-1002",
+                    "consignee": "TESLA MOTORS",
+                    "shipper": "PANASONIC",
+                    "description": "LITHIUM ION BATTERIES",
+                    "qty": "5000",
+                    "qty_unit": "KGS",
+                    "port_of_lading": "OSAKA",
+                    "port_of_unlading": "SAN FRANCISCO",
+                    "date": simulated_date_2
+                }
+            ]
+        }
+
+    async def parse(self, raw_data: Dict[str, Any]) -> List[Dict[str, Any]]:
+        # 从原始响应中提取需要保存的列表
+        return raw_data.get("results", [])
+
+    async def normalize(self, raw: RawTradeRecord) -> StandardTradeRecord:
+        data = raw.raw_json
+        
+        # 模拟企业名基础清洗
+        importer = data.get("consignee", "").strip().upper()
+        exporter = data.get("shipper", "").strip().upper()
+        
+        # 生成全局业务主键
+        record_id = str(uuid.uuid4())
+        
+        return StandardTradeRecord(
+            record_id=record_id,
+            source_record_id=raw.id, # 血缘追踪绑定
+            batch_no=raw.batch_no,
+            source_country=self.country_code,
+            trade_direction="import",
+            trade_date=datetime.strptime(data.get("date"), "%Y-%m-%d"),
+            importer_name=importer,
+            exporter_name=exporter,
+            product_name=data.get("description"),
+            weight=float(data.get("qty", 0)),
+            weight_unit=data.get("qty_unit"),
+            departure_port=data.get("port_of_lading"),
+            arrival_port=data.get("port_of_unlading")
+        )