feat: 接入巴西真实海关数据源并优化回补逻辑与HTTP客户端

- 实现巴西Comex Stat官方开放CSV数据源的拉取、解析与标准化流程，替换原Mock模拟数据
- 优化回补脚本run_backfill.py，支持巴西按年月范围批量执行回补任务
- 修复HTTP客户端SSL证书校验失败的降级重试逻辑，新增代理与SSL参数配置能力
- 添加项目下一步开发规划文档与测试日志示例

apps/worker/run_backfill.py

@@ -28,11 +28,21 @@ async def run_backfill(country_code: str, start_date: str, end_date: str):
     end_dt = datetime.strptime(end_date, "%Y-%m-%d")
     
     async with AsyncSessionLocal() as session:
+        # 巴西真实源支持按月范围回补，直接一次性下发给 connector。
+        if country_code == "BR":
+            connector = BrazilComexStatConnector(
+                session=session,
+                start_period=start_dt.strftime("%Y-%m"),
+                end_period=end_dt.strftime("%Y-%m"),
+            )
+            connector.batch_no = f"BACKFILL_{country_code}_{start_dt.strftime('%Y%m')}_{end_dt.strftime('%Y%m')}"
+            await connector.run()
+            app_logger.info(f"--- Finished BACKFILL Job for {country_code} ---")
+            return
+
         # 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:

docs/下一步开发规划与数据扩充方案.md

@@ -0,0 +1,58 @@
+# 下一步开发规划与数据扩充方案
+
+## 1. 现状评估
+
+根据当前项目代码和文档的实际情况，项目的核心架构（抓取 -> 原始留存 -> 标准化 -> 入库）和数据模型已经跑通。
+**关键状态**：目前所有国家（包括巴西、美国及扩展的 7 个国家）都在使用 Mock 模拟数据，尚未接入真实的外部网络请求。
+
+## 2. 哪个国家的数据最好拿？（国家接入优先级）
+
+从全球海关数据的开放度来看，拉美地区和部分官方统计源最容易获取，且合规风险最低。
+
+### 第一梯队（最好拿，官方全量开放，字段全）
+建议作为真实数据首个突破口。
+*   **巴西 (Brazil)**：官方的 Comex Stat 平台提供极高透明度的数据，可以直接下载包含明细的 CSV 文件或调用公开 API。
+*   **其他拉美国家 (如智利、哥伦比亚、秘鲁)**：普遍在政府公开网站提供进出口详细数据下载，字段完整度高。
+
+### 第二梯队（数据结构好，但缺少部分敏感信息）
+*   **欧盟 (Eurostat) / 英国**：官方 API 非常规范，数据极度稳定，适合做宏观与商品流向分析，但**没有具体的企业名**。
+*   **墨西哥 (Mexico)**：官方 SIAVI 平台有数据包，但解析复杂（多为西班牙语编码压缩包）。
+
+### 第三梯队（难拿，官方封闭，需对抗反爬或购买第三方）
+*   **美国 (US)**：官方（CBP）只有汇总，真实的提单（B/L）数据需要从港口或第三方商业公司获取。
+*   **印度 (India) / 越南 (Vietnam)**：官方经常关闭企业名或加严格验证码，市面上的明细数据基本靠第三方平台（如 Zauba）内部渠道流出，抓取难度大、数据脏。
+
+## 3. 怎么拿到更全的数据？（数据获取策略）
+
+“全”分为**覆盖全**和**历史全**，建议采用“官方打底 + 第三方补全”的策略：
+
+1.  **历史全量靠“静态下载”，增量靠“API轮询”**：
+    对于巴西等国，不要用 API 去爬过去 10 年的数据。直接从官方网站下载历史全量的 CSV 数据包放入 S3 或本地，通过解析脚本做一次性历史回补（Backfill）。API 只用来做 T+1 或 T+7 的增量更新。
+2.  **汇总与明细分层获取**：
+    官方源（如美国 CBP）往往只提供商品、金额、重量等汇总维度。要拿带企业名的提单明细，必须对接商业第三方平台（如 PIERS、ImportGenius 等的接口或页面抓取）。
+3.  **应对反爬保证数据不漏**：
+    对于必须抓取网页的 B/C 类数据源，完善 `http_client.py` 中的动态代理池逻辑，使用高质量的住宅代理（Residential Proxies）避免被封 IP 导致数据缺失。
+
+## 4. 接下来的开发任务规划
+
+当前的首要任务是将系统从“Mock 跑通”推进到“真实数据落地”。开发任务及状态如下：
+
+- [ ] **1. 将巴西改为真实数据源（首个真实闭环）**
+  - 修改 `packages/connectors/brazil.py` 中的 `fetch` 方法。
+  - 对接巴西 Comex Stat 真实的 API 或自动下载其公开的 CSV 文件。
+  - 完成真实葡语字段到标准模型的映射映射与验证。
+
+- [ ] **2. 补全代理池与网络请求防封能力**
+  - 在 `packages/core/http_client.py` 中落实真实的代理 IP 获取逻辑（对接如快代理、BrightData 等）。
+  - 增加对 403 / 验证码 / Cloudflare 拦截的识别与重试机制。
+
+- [ ] **3. 实现真实的历史数据回补（Backfill）**
+  - 修改 `apps/worker/run_backfill.py`。
+  - 设计一套读取本地或 S3 上的 CSV 文件并批量走 `parse -> save_raw -> normalize` 的流水线，先把巴西过去 3 年的数据灌入数据库。
+
+- [ ] **4. 攻克第二个真实国家（建议智利或墨西哥）**
+  - 创建新的 Connector，复用巴西的成功经验，处理不同语言的字典映射。
+
+- [ ] **5. 引入真实的分布式任务调度**
+  - 当前通过 `asyncio.run()` 单进程跑脚本无法满足后续多国家并发和失败重试。
+  - 规划引入 Celery 或 Prefect，将各个国家的 `run()` 注册为定时任务（Cron）和死信重试任务。

logs/app_2026-06-10.log

@@ -0,0 +1,11 @@
+2026-06-10 22:39:48.907 | DEBUG    | packages.core.http_client:request:53 - Request: GET https://balanca.economia.gov.br/balanca/bd/comexstat-bd/ncm/EXP_2026.csv | Attempt: 1/3
+2026-06-10 22:39:50.652 | WARNING  | packages.core.http_client:request:77 - Request Error: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1081) for https://balanca.economia.gov.br/balanca/bd/comexstat-bd/ncm/EXP_2026.csv
+2026-06-10 22:39:52.707 | DEBUG    | packages.core.http_client:request:53 - Request: GET https://balanca.economia.gov.br/balanca/bd/comexstat-bd/ncm/EXP_2026.csv | Attempt: 2/3
+2026-06-10 22:39:54.324 | WARNING  | packages.core.http_client:request:77 - Request Error: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1081) for https://balanca.economia.gov.br/balanca/bd/comexstat-bd/ncm/EXP_2026.csv
+2026-06-10 22:39:58.422 | DEBUG    | packages.core.http_client:request:53 - Request: GET https://balanca.economia.gov.br/balanca/bd/comexstat-bd/ncm/EXP_2026.csv | Attempt: 3/3
+2026-06-10 22:40:00.345 | WARNING  | packages.core.http_client:request:77 - Request Error: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1081) for https://balanca.economia.gov.br/balanca/bd/comexstat-bd/ncm/EXP_2026.csv
+2026-06-10 22:40:08.353 | ERROR    | packages.core.http_client:request:82 - Max retries reached for https://balanca.economia.gov.br/balanca/bd/comexstat-bd/ncm/EXP_2026.csv. Last error: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1081)
+2026-06-10 22:40:49.505 | DEBUG    | packages.core.http_client:request:55 - Request: GET https://balanca.economia.gov.br/balanca/bd/comexstat-bd/ncm/EXP_2026.csv | Attempt: 1/3
+2026-06-10 22:40:51.104 | WARNING  | packages.core.http_client:request:79 - Request Error: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1081) for https://balanca.economia.gov.br/balanca/bd/comexstat-bd/ncm/EXP_2026.csv
+2026-06-10 22:40:51.104 | WARNING  | packages.core.http_client:request:82 - SSL 证书校验失败，降级为 verify=False 重试: https://balanca.economia.gov.br/balanca/bd/comexstat-bd/ncm/EXP_2026.csv
+2026-06-10 22:40:51.106 | DEBUG    | packages.core.http_client:request:55 - Request: GET https://balanca.economia.gov.br/balanca/bd/comexstat-bd/ncm/EXP_2026.csv | Attempt: 2/3

packages/connectors/brazil.py

@@ -1,131 +1,193 @@
+import csv
+import io
 import uuid
-from typing import List, Dict, Any
-from datetime import datetime
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Optional
+
 from packages.connectors.base import BaseConnector
+from packages.core.http_client import http_client
 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
+from packages.dictionaries.brazil import TRANSPORT_MODE_MAP
 
 class BrazilComexStatConnector(BaseConnector):
     """
-    巴西 Comex Stat 官方统计数据接入适配器
-    这是一个基于 API / CSV 的真实数据源结构模拟
+    巴西 Comex Stat 官方开放数据适配器。
+    当前优先接入官方年度 CSV，并按月过滤出真实记录。
     """
     country_code = "BR"
-    source_system = "COMEX_STAT_API"
-    parser_version = "v1.1.0"
+    source_system = "COMEX_STAT_OPEN_DATA"
+    parser_version = "v1.2.0"
+    BASE_DATA_URL = "https://balanca.economia.gov.br/balanca/bd/comexstat-bd/ncm"
+
+    def __init__(
+        self,
+        session,
+        start_period: Optional[str] = None,
+        end_period: Optional[str] = None,
+        max_rows_per_slice: Optional[int] = None,
+    ):
+        super().__init__(session=session)
+        self.start_period = start_period
+        self.end_period = end_period
+        self.max_rows_per_slice = max_rows_per_slice
 
     async def discover(self) -> List[Any]:
-        # 模拟发现 2024 年 5 月的数据批次
-        return [{"year": 2024, "month": 5}]
+        periods = self._build_periods()
+        task_slices: List[Dict[str, Any]] = []
+
+        for year, month in periods:
+            task_slices.append({"year": year, "month": month, "trade_direction": "export"})
+            task_slices.append({"year": year, "month": month, "trade_direction": "import"})
+
+        return task_slices
 
     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")
-        
+        year = int(task_slice["year"])
+        month = int(task_slice["month"])
+        trade_direction = task_slice["trade_direction"]
+        file_prefix = "EXP" if trade_direction == "export" else "IMP"
+        url = f"{self.BASE_DATA_URL}/{file_prefix}_{year}.csv"
+
+        response = await http_client.get(
+            url,
+            headers={"User-Agent": "customs-data/1.0"},
+            timeout=180,
+            follow_redirects=True,
+            allow_insecure_fallback=True,
+        )
+
+        text = response.content.decode("latin-1")
+        reader = csv.DictReader(io.StringIO(text), delimiter=";")
+        rows: List[Dict[str, Any]] = []
+
+        for row in reader:
+            if not row:
+                continue
+
+            row_month = self._safe_int(row.get("CO_MES"))
+            if row_month != month:
+                continue
+
+            rows.append(
+                {
+                    "co_ano": row.get("CO_ANO"),
+                    "co_mes": row.get("CO_MES"),
+                    "co_ncm": row.get("CO_NCM"),
+                    "co_unid": row.get("CO_UNID"),
+                    "co_pais": row.get("CO_PAIS"),
+                    "sg_uf_ncm": row.get("SG_UF_NCM"),
+                    "co_via": row.get("CO_VIA"),
+                    "co_urf": row.get("CO_URF"),
+                    "qt_estat": row.get("QT_ESTAT"),
+                    "kg_liquido": row.get("KG_LIQUIDO"),
+                    "vl_fob": row.get("VL_FOB"),
+                    "trade_direction": trade_direction,
+                    "source_file_url": url,
+                }
+            )
+
+            if self.max_rows_per_slice and len(rows) >= self.max_rows_per_slice:
+                break
+
         return {
             "metadata": {
                 "status": "success",
-                "period": f"{task_slice['year']}-{task_slice['month']:02d}"
+                "period": f"{year}-{month:02d}",
+                "trade_direction": trade_direction,
+                "source_file_url": url,
             },
-            "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"
-                }
-            ]
+            "data": rows,
         }
 
     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):
+        if isinstance(raw_data, list):
             return raw_data
-        else:
-            return []
+        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())
-        
+        trade_direction = data.get("trade_direction", "export")
+        trade_date = datetime(
+            year=self._safe_int(data.get("co_ano")) or 1970,
+            month=self._safe_int(data.get("co_mes")) or 1,
+            day=1,
+        )
+        ncm_code = (data.get("co_ncm") or "").strip()
+        partner_country = self._truncate(data.get("co_pais"), 10)
+        transport_mode = TRANSPORT_MODE_MAP.get((data.get("co_via") or "").strip().upper(), "UNKNOWN")
+        record_id = str(uuid.uuid5(uuid.NAMESPACE_URL, f"{self.country_code}:{raw.id}"))
+
         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
+            trade_date=trade_date,
+            importer_name=None,
+            exporter_name=None,
+            hs_code=ncm_code,
+            product_name=f"NCM {ncm_code}" if ncm_code else None,
+            amount=self._safe_float(data.get("vl_fob")),
+            currency="USD",
+            weight=self._safe_float(data.get("kg_liquido")),
+            weight_unit="KG",
+            origin_country=partner_country if trade_direction == "import" else self.country_code,
+            destination_country=partner_country if trade_direction == "export" else self.country_code,
+            departure_port=self._truncate(data.get("co_urf"), 100) if trade_direction == "export" else None,
+            arrival_port=self._truncate(data.get("co_urf"), 100) if trade_direction == "import" else None,
+            transport_mode=transport_mode,
         )
+
+    def _build_periods(self) -> List[tuple[int, int]]:
+        if self.start_period and self.end_period:
+            return self._month_range(self.start_period, self.end_period)
+
+        now = datetime.now(timezone.utc)
+        year = now.year
+        month = now.month - 1
+        if month == 0:
+            year -= 1
+            month = 12
+        return [(year, month)]
+
+    def _month_range(self, start_period: str, end_period: str) -> List[tuple[int, int]]:
+        start = datetime.strptime(start_period, "%Y-%m")
+        end = datetime.strptime(end_period, "%Y-%m")
+        periods: List[tuple[int, int]] = []
+        current = start
+
+        while current <= end:
+            periods.append((current.year, current.month))
+            if current.month == 12:
+                current = current.replace(year=current.year + 1, month=1)
+            else:
+                current = current.replace(month=current.month + 1)
+
+        return periods
+
+    @staticmethod
+    def _safe_float(value: Optional[str]) -> Optional[float]:
+        if value in (None, ""):
+            return None
+        try:
+            return float(str(value).replace(",", "."))
+        except (TypeError, ValueError):
+            return None
+
+    @staticmethod
+    def _safe_int(value: Optional[str]) -> Optional[int]:
+        if value in (None, ""):
+            return None
+        try:
+            return int(str(value).strip())
+        except (TypeError, ValueError):
+            return None
+
+    @staticmethod
+    def _truncate(value: Optional[str], max_length: int) -> Optional[str]:
+        if not value:
+            return None
+        return str(value).strip()[:max_length]

packages/core/http_client.py

@@ -39,14 +39,19 @@ class BaseHttpClient:
         
         attempt = 0
         last_exception = None
+        verify = kwargs.pop("verify", True)
+        allow_insecure_fallback = kwargs.pop("allow_insecure_fallback", False)
         
         while attempt < self.max_retries:
             attempt += 1
             proxy = await self._get_proxy()
-            proxies = {"http://": proxy, "https://": proxy} if proxy else None
             
             try:
-                async with httpx.AsyncClient(proxies=proxies, timeout=self.timeout) as client:
+                client_kwargs = {"timeout": self.timeout, "verify": verify}
+                if proxy:
+                    client_kwargs["proxy"] = proxy
+
+                async with httpx.AsyncClient(**client_kwargs) as client:
                     app_logger.debug(f"Request: {method} {url} | Attempt: {attempt}/{self.max_retries}")
                     response = await client.request(
                         method=method,
@@ -73,6 +78,10 @@ class BaseHttpClient:
             except httpx.RequestError as e:
                 app_logger.warning(f"Request Error: {str(e)} for {url}")
                 last_exception = e
+                if allow_insecure_fallback and verify and "CERTIFICATE_VERIFY_FAILED" in str(e):
+                    app_logger.warning(f"SSL 证书校验失败，降级为 verify=False 重试: {url}")
+                    verify = False
+                    continue
                 await asyncio.sleep(2 ** attempt)
                 continue