Load data directly from CSV using pandas

apps/api/routers/trade.py

@@ -13,41 +13,75 @@ router = APIRouter()
 async def search_trade_records(
     query: TradeQueryRequest,
     db: AsyncSession = Depends(get_db_session)
-):    # 直接返回mock数据，展示效果
+):
+    """
+    检索标准贸易记录 - 直接从CSV加载数据
+    """
     from datetime import datetime
-    mock_items = [
-        TradeRecordResponse(
-            record_id=f"MOCK-{i}",
-            source_record_id=f"SRC-{i}",
-            source_country="US" if i % 2 == 0 else "CN",
-            trade_direction="import" if i % 2 == 0 else "export",
-            trade_date=datetime(2023, 1, i+1),
-            importer_name=f"进口商{i}",
-            exporter_name=f"出口商{i}",
-            hs_code=f"847130{i:02d}",
-            product_name=f"电子产品 - 样品{i}",
-            amount=10000.0 + i * 1000,
-            currency="USD",
-            weight=500.0 + i * 10,
-            weight_unit="KG",
-            origin_country="CN",
-            destination_country="US",
-            departure_port="上海港",
-            arrival_port="洛杉矶港",
-            transport_mode="SEA"
-        )
-        for i in range(min(query.limit, 20))
-    ]
+    import pandas as pd
+    from pathlib import Path
     
-    return PaginatedResponse(
-        total=10000,
-        items=mock_items
-    )    """
-    检索标准贸易记录
-    支持按国家、方向、HS、商品、企业名、起运/目的国及日期范围筛选
+    # 加载CSV数据
+    csv_path = Path(__file__).parent.parent.parent.parent / "data" / "standard_trade_records_sample.csv"
+    df = pd.read_csv(csv_path)
     
-    当有全文搜索需求（product_name/importer_name/exporter_name）时，优先使用 Elasticsearch
-    """
+    # 应用过滤条件
+    filtered_df = df.copy()
+    
+    if query.source_country:
+        filtered_df = filtered_df[filtered_df['source_country'] == query.source_country]
+    if query.trade_direction:
+        filtered_df = filtered_df[filtered_df['trade_direction'] == query.trade_direction]
+    if query.hs_code:
+        filtered_df = filtered_df[filtered_df['hs_code'].astype(str).str.startswith(str(query.hs_code))]
+    if query.product_name:
+        filtered_df = filtered_df[filtered_df['product_name'].str.contains(query.product_name, na=False, case=False)]
+    if query.importer_name:
+        filtered_df = filtered_df[filtered_df['importer_name'].str.contains(query.importer_name, na=False, case=False)]
+    if query.exporter_name:
+        filtered_df = filtered_df[filtered_df['exporter_name'].str.contains(query.exporter_name, na=False, case=False)]
+    
+    total = len(filtered_df)
+    
+    # 分页
+    offset = (query.page - 1) * query.limit
+    page_df = filtered_df.iloc[offset:offset + query.limit]
+    
+    # 转换为响应格式
+    items = []
+    for _, row in page_df.iterrows():
+        # 处理日期 - 只取日期部分
+        trade_date_str = str(row['trade_date']).split()[0] if pd.notna(row['trade_date']) else '2023-01-01'
+        try:
+            trade_date = datetime.fromisoformat(trade_date_str)
+        except:
+            trade_date = datetime(2023, 1, 1)
+        
+        items.append(TradeRecordResponse(
+            record_id=str(row['record_id']),
+            source_record_id=str(row['source_record_id']),
+            source_country=str(row['source_country']),
+            trade_direction=str(row['trade_direction']),
+            trade_date=trade_date,
+            importer_name=str(row['importer_name']) if pd.notna(row['importer_name']) else "",
+            exporter_name=str(row['exporter_name']) if pd.notna(row['exporter_name']) else "",
+            hs_code=str(row['hs_code']) if pd.notna(row['hs_code']) else None,
+            product_name=str(row['product_name']) if pd.notna(row['product_name']) else None,
+            amount=float(row['amount']) if pd.notna(row['amount']) else None,
+            currency=str(row['currency']) if pd.notna(row['currency']) else None,
+            weight=float(row['weight']) if pd.notna(row['weight']) else None,
+            weight_unit=str(row['weight_unit']) if pd.notna(row['weight_unit']) else None,
+            origin_country=str(row['origin_country']) if pd.notna(row['origin_country']) else None,
+            destination_country=str(row['destination_country']) if pd.notna(row['destination_country']) else None,
+            departure_port=str(row['departure_port']) if pd.notna(row['departure_port']) else None,
+            arrival_port=str(row['arrival_port']) if pd.notna(row['arrival_port']) else None,
+            transport_mode=str(row['transport_mode']) if pd.notna(row['transport_mode']) else None
+        ))
+    
+    return PaginatedResponse(
+        total=total,
+        items=items
+    )
     print(f"[DEBUG] Received query: {query}")
     
     # 判断是否需要全文搜索

requirements.txt

@@ -16,3 +16,4 @@ boto3==1.34.131
 loguru==0.7.2
 aiohttp>=3.9.5
 pytest==9.0.3
+pandas>=2.0.0