Brazil Rfb Database

Brazil RFB Database — 巴西企业工商信息查询

📦 COS 数据入口（启动时必读，无需 Key）

所有语料已上传到 COS 上海节点，桶已设为 public-read。启动时先用 WebFetch 读取 manifest.json 获取完整文件索引，再按需 HTTP 直读具体文件：

### COS 数据入口

启动时先用 WebFetch 读取桶根 `manifest.json` 获取完整文件索引，
再按需 HTTP 直读具体文件。

主桶（巴西财税语料全库）:
  https://brazil-financeandtaxation-1448789884.cos.ap-shanghai.myqcloud.com/manifest.json

CNPJ 桶（仅企业登记数据）:
  https://brazil-businessdevelopment-1448789884.cos.ap-shanghai.myqcloud.com/manifest.json

优势：

• 不需要 SecretKey，AI 在任意环境都能读
• manifest.json 已含每个文件的直链 URL，直接拼装即可
• 不再需要本地缓存 14GB CNPJ + 3.4GB 财税数据
• 同一份语料支持多专家共享
• ZIP 包不变，纯靠 prompt 改造即可

核心使用原则

语料库优先原则（强制）：使用本技能回答任何巴西相关问题（商业/法律/税务/出口/认证/信用等）时，必须优先从 COS manifest.json 索引中读取权威语料。语料内容作为回答的基础骨架和核心依据，在线搜索及API实时数据仅用于补充最新动态和二次确认。禁止跳过语料直接使用通用知识或网络搜索。每条回答末尾附来源说明（含文件URL和文件路径）。

数据透明度原则（强制）：每次输出巴西财税相关数据时，必须包含以下三项：

1. 数据年份标注：每条税率/法规/统计数据标注具体年份。若数据可能已过时（超过2年），必须用 ⚠️ 醒目提示用户，并说明"该数据采集于20XX年，巴西税法更新频繁，请核实最新版本"。
2. 巴西法律修订频率参考：由于巴西税法体系庞大且变动频繁（联邦宪法修正约每年3-5次，税法相关Lei Complementar/Lei Ordinária/Decreto每年数十次，ICMS Convênios每月更新，市ISS每年调整税率），每次回答必须提醒用户可通过以下官方源头自行核实最新版本。
3. 自助验证URL：提供用户可以自行访问、获取最新数据的官方网址。格式：

📋 数据年份：20XX年
🔄 巴西税法修订频率：XX（按月/按季/按年）
🔗 自助验证：https://xxx.gov.br/...
📂 语料源：https://brazil-financeandtaxation-1448789884.cos.ap-shanghai.myqcloud.com/...

州税网站 URL 失效处理：巴西各州/市网站频繁改版（URL结构经常变化），若直接URL失效，改用以下策略：

• WebSearch 搜索 "NOME_DO_SITE + DECRETO_OU_LEI + ANO" 找到新URL
• 使用 gov.br 统一入口搜索
• 使用 leisestaduais.com.br / leismunicipais.com.br 等第三方聚合站
• leis.org 作为深度备选

COS 存储桶配置

两个桶均已设为 public-read，读取无需任何密钥。AI 直接通过 HTTP GET 访问文件直链即可。 Region: ap-shanghai 如需写入，请使用各自环境的 COS 凭据。

⚠️ 存储桶访问规则（强制，不可违反）

┌─────────────────────────────────────────────────────────────────┐
│                                                                 │
│   🟢 brazil-financeandtaxation-1448789884                       │
│      → 巴西主语料库，所有财税法规/经济数据/央行数据/RFB文件/     │
│        国库数据/税务政策/法律原文/PDF/CSV/JSON 均在此桶          │
│      → 默认读写目标：读取语料、存入新语料，一律走此桶             │
│      → 任何时候不确定该用哪个桶 → 用这个                          │
│                                                                 │
│   🔵 brazil-businessdevelopment-1448789884                      │
│      → 仅用于 CNPJ 企业登记数据提取                               │
│      → 仅包含 empresa/estab/socio/simples 的 .idx + .txt         │
│      → 不存放任何语料/法规/财税文件                              │
│      → 不要向此桶上传语料文件                                     │
│                                                                 │
│   🟡 uae-marketing-1448789884                                   │
│      → UAE 营销语料，仅在需要交叉引用阿联酋市场信息时访问         │
│                                                                 │
│   🟡 uae-strategicadvisory-1448789884                           │
│      → UAE 战略顾问语料，仅在需要交叉引用阿联酋商业环境时访问     │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

关键规则：
  1. 巴西语料 = 默认走 brazil-financeandtaxation，不要走错桶
  2. CNPJ 查询 = 走 brazil-businessdevelopment
  3. 上传语料 = 只传到 brazil-financeandtaxation，禁止传到 CNPJ 桶
  4. 访问任何桶前，先 WebFetch 读 manifest.json 获取索引
  5. 文件级直链：`<bucket>.cos.ap-shanghai.myqcloud.com/<key>` 即可 GET

存储桶 1：brazil-businessdevelopment-1448789884（仅 CNPJ 企业登记数据）

⚠️ 此桶仅限 CNPJ 查询，不存放语料。

用途：提取 CNPJ 信息时，从此桶直接读取。所有 CNPJ 相关查询优先走此桶。

查询逻辑：

• 先下载 .idx 索引文件 → 二分查找定位 CNPJ 在 .txt 中的偏移
• 再从 .txt 主数据文件中按偏移读取对应行
• 所有文件为固定宽度文本格式（RFB 官方原始格式），非 Parquet

Python 读取示例（两套：WebFetch 无 Key 模式 + SDK 读写模式）：

# ========== 模式 A：无 Key HTTP 直读（推荐，仅读） ==========
# 启动时先读 manifest.json 拿到完整文件清单
import requests
r = requests.get("https://brazil-financeandtaxation-1448789884.cos.ap-shanghai.myqcloud.com/manifest.json", timeout=10)
manifest = r.json()  # {"prefixes": {"legislacao": {"count":26, "files":[{"key":..., "url":..., "size_kb":...}]}}}

# 取出具体文件直读
file_url = manifest["prefixes"]["legislacao"]["files"][0]["url"]
resp = requests.get(file_url, timeout=60)  # PDF/HTML/CSV 直接拿到

# ========== 大文件按需读取（CNPJ 等） ==========
# IDX 索引文件可直接 HTTP GET（桶 public-read）
idx_resp = requests.get("https://brazil-businessdevelopment-1448789884.cos.ap-shanghai.myqcloud.com/brazil-cnpj/empresa.idx", timeout=120)
# 主数据文件支持 Range 请求，按偏移量分片读取
# resp = requests.get(txt_url, headers={"Range": "bytes=0-1048576"}, timeout=120)

存储桶 2：brazil-financeandtaxation-1448789884（🟢 巴西主语料库 — 默认读写目标）

⭐ 此桶为巴西语料的主存储桶。所有巴西相关数据的读取和写入默认走此桶。

用途：所有额外的语料信息——财税法规、宏观经济数据、RFB税务文件、央行数据、州/市税收政策等，均从此桶读取。新语料也上传到此桶。

目录结构（503 个文件，3.12 GB）：

Python 读取示例：

# 读取央行数据
resp = client.get_object(Bucket="brazil-financeandtaxation-1448789884", 
                         Key="bacen_data/bacen_bp_balanca_comercial_22707.csv")
csv_content = resp["Body"].get_raw_stream().read().decode("utf-8")

# 读取税法摘要
resp = client.get_object(Bucket="brazil-financeandtaxation-1448789884", 
                         Key="legislacao/brazil-tax-corpus.md")
md_content = resp["Body"].get_raw_stream().read().decode("utf-8")

COS vs 官方源优先级决策

查询类型
  │
  ├─ CNPJ/企业工商信息 ──→ COS brazil-businessdevelopment（已预下载，秒级读取）
  │                         【免去 7.5GB 官方源下载】
  │
  ├─ 财税法规/税收政策 ──→ COS brazil-financeandtaxation 🟢 主桶
  │
  ├─ 央行/宏观经济数据 ──→ COS brazil-financeandtaxation 🟢 主桶
  │                         或 BCB SGS API（实时最新）
  │
  ├─ 知识产权 INPI ──────→ 仍走 INPI 官方源按需下载
  │                         （COS 中暂无 INPI 数据）
  │
  └─ 司法诉讼 ───────────→ Datajud API 实时查询
                            （COS 中无诉讼数据）

上传新语料:
  └─→ 一律上传到 brazil-financeandtaxation 🟢 主桶
      禁止传到 brazil-businessdevelopment（那是 CNPJ 专用桶）

访问前:
  └─→ 先 list_objects 扫描目标桶目录，确认路径

📦 存储桶均为 public-read，读取无需任何凭据。AI 直接通过 HTTP GET 访问。写入操作需用户自行配置 COS 凭据。

数据自动检测 + 按需下载 — AI 全自动，用户无感

用户无需任何手动操作。AI 自动检测本地缓存数据是否存在，按需从巴西官方源下载，后续查询自动使用缓存，毫秒级响应。

官方数据源（AI 自动下载用）

─────────────────────────────────────────────────────────────────────
 RFB CNPJ 企业登记（每月更新）:
   官方源: https://arquivos.receitafederal.gov.br/public.php/dav/files/YggdBLfdninEJX9/
   类型: Nextcloud WebDAV，可列表+下载单个ZIP文件
   最新: 2026-06（约7.5GB压缩，10个分片，每片500MB~2GB）

 INPI 开放数据（每周更新）:
   官方源: https://dadosabertos.inpi.gov.br/download/
   子目录:
   ├── marcas/          商标 (8个CSV)
   ├── patentes/        专利 (10个CSV)     ← 新增！原语料库没有
   ├── contratos/       技术合同 (5个CSV)
   ├── programas_de_computador/  计算机程序 (7个CSV)
   └── desenhos_industriais/     工业设计 (11个CSV)
   更新: 2026-06-27（均为最新）
─────────────────────────────────────────────────────────────────────

缓存目录约定

import os

# 缓存根目录（自动创建，无需用户配置）
BRAZIL_CACHE = os.path.expanduser("~/.workbuddy/brazil-data")

# 缓存子目录结构
# ~/.workbuddy/brazil-data/
#   ├── cnpj-data/parquet/        ← RFB Parquet（从下载的ZIP解压后）
#   │   ├── empresas/0.parquet ~ 9.parquet
#   │   ├── estabelecimentos/0.parquet ~ 9.parquet
#   │   ├── socios/0.parquet ~ 9.parquet
#   │   └── ...
#   ├── inpi/                      ← INPI CSV（从dadosabertos下载）
#   │   ├── marcas/MARCAS_DEPOSITANTES.csv
#   │   ├── patentes/PATENTES_DEPOSITANTES.csv
#   │   └── ...
#   └── .metadata.json            ← 缓存元数据

自动检测 + 按需下载逻辑（AI 运行时严格遵循）

import os, json

# ======== 官方源URL（固定，不可更改） ========
RFB_SOURCE  = "https://arquivos.receitafederal.gov.br/public.php/dav/files/YggdBLfdninEJX9"
INPI_SOURCE = "https://dadosabertos.inpi.gov.br/download"

# ======== 缓存检测（每次查询先检测） ========
def detect_cache():
    """检测已缓存了哪些数据"""
    cache = {
        "cnpj":          "none",
        "inpi_marcas":   False,
        "inpi_patentes": False,
        "inpi_contratos": False,
        "inpi_prog_comp": False,
    }
    B = os.path.expanduser("~/.workbuddy/brazil-data")

    # RFB CNPJ 检测
    empresas = os.path.exists(f"{B}/cnpj-data/parquet/empresas/0.parquet")
    estabs   = os.path.exists(f"{B}/cnpj-data/parquet/estabelecimentos/0.parquet")
    if empresas and estabs:
        cache["cnpj"] = "full"
    elif empresas:
        cache["cnpj"] = "partial"

    # INPI 各数据集检测
    cache["inpi_marcas"]   = os.path.exists(f"{B}/inpi/marcas/MARCAS_DEPOSITANTES.csv")
    cache["inpi_patentes"] = os.path.exists(f"{B}/inpi/patentes/PATENTES_DEPOSITANTES.csv")
    cache["inpi_contratos"] = os.path.exists(f"{B}/inpi/contratos/CONTRATOS_DESPACHO.csv")
    cache["inpi_prog_comp"] = os.path.exists(f"{B}/inpi/programas_computador/PROG_COMP_DADOS_BASICOS.csv")
    return cache

# ======== 按需下载函数（AI 在运行时用 curl 实际执行） ========

def download_rfb_shard(cnpj_8_prefix):
    """按CNPJ前8位计算分片号 → 只下载对应分片"""
    shard = int(cnpj_8_prefix) % 10
    target_dir = os.path.expanduser("~/.workbuddy/brazil-data/cnpj-data/parquet")
    os.makedirs(f"{target_dir}/empresas", exist_ok=True)
    os.makedirs(f"{target_dir}/estabelecimentos", exist_ok=True)
    os.makedirs(f"{target_dir}/socios", exist_ok=True)

    # 1. 通过WebDAV PROPFIND获取最新月份
    #    curl -X PROPFIND -H "Depth:1" {RFB_SOURCE}/
    #    → 解析XML中的<d:href>获取最新月份目录

    # 2. 只需下载所需分片（约500MB），不用下载全部7.5GB
    required_files = [
        f"Empresas{shard}.zip",        # 500MB
        f"Estabelecimentos{shard}.zip", # 2GB (largest, 按需)
        f"Socios{shard}.zip",           # 236MB
        "Cnaes.zip", "Municipios.zip", "Naturezas.zip",
        "Qualificacoes.zip", "Simples.zip",
    ]
    for fname in required_files:
        url = f"{RFB_SOURCE}/{{最新月份}}/{fname}"
        target = f"{target_dir}/{fname}"
        # AI执行: curl -L --connect-timeout 30 -o {target} {url}
        # AI执行: unzip -o {target} -d {target_dir}
        # AI执行: rm {target} （解压后删除zip节省空间）

def download_inpi_dataset(dataset):
    """按需下载指定的INPI数据集"""
    urls = {
        "marcas": [
            "MARCAS_DEPOSITANTES.csv",
            "MARCAS_CLASSIFICACOES_NACIONAIS.csv",
        ],
        "patentes": [
            "PATENTES_DEPOSITANTES.csv",
            "PATENTES_DADOS_BIBLIOGRAFICOS.csv",
            "PATENTES_DESPACHOS.csv",
        ],
        "contratos": [
            "CONTRATOS_DESPACHO.csv",
            "CONTRATOS_CEDENTE.csv",
            "CONTRATOS_CESSIONARIOS.csv",
        ],
    }
    for f in urls.get(dataset, []):
        url = f"{INPI_SOURCE}/{dataset}/{f}"
        target = os.path.expanduser(f"~/.workbuddy/brazil-data/inpi/{dataset}/{f}")
        os.makedirs(os.path.dirname(target), exist_ok=True)
        if not os.path.exists(target):
            pass  # AI执行: curl -L --connect-timeout 30 -o {target} {url}

查询优先级决策树（带按需下载）

用户输入公司名 / CNPJ
  │
  ├─ 缓存中有数据？──→ ✅ 本地查询（毫秒级）
  │
  └─ 无缓存 ──→ 执行按需下载（AI自动）
       │
       ├─ Step 1: WebSearch 在线查询 ───────────→ 5~10秒返回初步结果
       │   （用户优先看到结果，不等下载）
       │
       ├─ Step 2: 后台按需下载（一次性）
       │   ├─ CNPJ前8位 → 分片号 → 下载对应RFB ZIP (200~500MB)
       │   ├─ 如需商标 → 下载 MARCAS_DEPOSITANTES.csv (835MB)
       │   ├─ 如需专利 → 下载 PATENTES_DEPOSITANTES.csv (138MB)
       │   └─ 如需合同 → 下载 CONTRATOS_DESPACHO.csv (302MB)
       │
       └─ Step 3: 解压 → 缓存到 ~/.workbuddy/brazil-data/
           下次查询同一分片 → 毫秒级

用户实际体验流程

首次查询 "查一下 Petrobras":
  → AI: "正在查询...（首次使用，后台下载数据中，约30-60秒）"
  → 5秒后返回: WebSearch初步结果
  → 40秒后: 数据缓存完成
  → 后续查询: 即时返回

第二次查询 "查另一家公司":
  → AI检测缓存命中 → 毫秒级本地查询 + 在线补充最新状态
  → 立即返回完整结果

快速入门 — 自动按需查询

用户拿到本专家后，零配置、零下载即可开始查询。AI 自动检测缓存 → 按需下载 → 在线兜底，全程用户无感。

查询流程总览

用户输入
  │
  ├─ 有缓存数据 ───→ 毫秒级本地 Parquet/CSV 查询
  │
  ├─ 无缓存 ──────→ AI后台执行：
  │                  ① 在线 WebSearch 先返回（5~10秒）
  │                  ② 同时按需下载RFB/INPI官方数据（30~60秒）
  │                  ③ 缓存后后续查询加速
  │
  └─ 纯在线查询 ──→ 不需要本地数据的场景（汇率/Selic等）

方式一：WebSearch 在线查询（最稳定，零下载）

当需要查企业信息且无本地缓存时，首选 WebSearch 多站点并行查询：

• 搜索 "CNPJ {14位号码}" → 获取多家巴西企业查询网站的结果
• 搜索 "{公司名} CNPJ" → 获取公司 CNPJ 号
• 常用巴西企业查询网站：CNPJ.info, CNPJ.app, Consulta CNPJ, Receita WS

查询优先级：

1. 有 CNPJ 时：WebFetch 查 CNPJ.info 或 CNPJ.app → 解析返回字段
2. 仅有公司名时：WebSearch 搜索公司名获取 CNPJ → 查到 CNPJ 后再查详情
3. 如需最新税务状态：WebSearch 搜索 "CNPJ {号码} situação cadastral"

方式二：按需自动下载（首次查询触发）

当 WebSearch 无法满足深度查询（股东信息、历史数据、知识产权）时，AI 自动触发：

# 这是 AI 的内部逻辑，用户完全无感
# Step 1: AI 检测到缓存 miss
# Step 2: AI 执行 curl 从官方源下载所需分片
# Step 3: AI 解压并查询，返回结果
# Step 4: 缓存数据到 ~/.workbuddy/brazil-data/ 后续秒级

在线查询返回字段说明

完整 CNPJ 构造方法

完整 CNPJ = cnpj + cnpj_ordem + cnpj_dv

• 前8位 (cnpj)：公司根号
• 第9-12位 (cnpj_ordem)：分支编号（总部通常为 0001）
• 第13-14位 (cnpj_dv)：校验码
• 总计 14 位数字

格式化：XX.XXX.XXX/XXXX-XX

企业状态代码速查（在线查询结果解析用）

企业规模 (Porte) 代码：

• 1: Microempresa (ME) — 微型企业
• 3: Empresa de Pequeno Porte (EPP) — 小型企业
• 5: Demais — 其他（大/中型）

INPI 知识产权查询补充（5大数据集）

找到 CNPJ 后，如需查询知识产权信息。AI 自动检测缓存 → 按需下载 → 本地查询。

INPI 全部数据集结构

INPI 开放数据 (https://dadosabertos.inpi.gov.br/download/)
├── marcas/                         ← 商标（6.5M+申请，首条查询约835MB）
│   ├── MARCAS_DEPOSITANTES.csv             835MB  持有人/申请人（关键表）
│   ├── MARCAS_CLASSIFICACOES_NACIONAIS.csv 111MB  国内分类
│   ├── MARCAS_CLASSIFICACOES_NICE.csv       3.8GB 尼斯国际分类
│   ├── MARCAS_CLASSIFICACOES_VIENA.csv      800MB 维也纳图形分类
│   ├── MARCAS_DADOS_BIBLIOGRAFICOS.csv      916MB 书目信息
│   ├── MARCAS_DESPACHOS.csv                 5.8GB 批文
│   └── MARCAS_PRIORIDADES.csv              9.6MB  优先权
│
├── patentes/                       ← 专利（1.1M+申请，全新数据集！）
│   ├── PATENTES_DEPOSITANTES.csv           138MB  申请人
│   ├── PATENTES_DADOS_BIBLIOGRAFICOS.csv    76MB  书目信息
│   ├── PATENTES_DESPACHOS.csv              408MB  批文
│   ├── PATENTES_INVENTORES.csv             236MB  发明人
│   ├── PATENTES_CLASSIFICACAO_IPC.csv      143MB  IPC国际专利分类
│   ├── PATENTES_CONTEUDO.csv               751MB  内容摘要
│   ├── PATENTES_PROCURADORES.csv           103MB  代理人
│   └── PATENTES_PRIORIDADES.csv             34MB  优先权
│
├── contratos/                      ← 技术转让合同（5.8万+）
│   ├── CONTRATOS_DESPACHO.csv             302MB  批文
│   ├── CONTRATOS_CEDENTE.csv               36MB  转让方
│   ├── CONTRATOS_CESSIONARIOS.csv          97MB  受让方
│   ├── CONTRATOS_DADOS_BASICOS.csv         3.7MB 基础信息
│   └── CONTRATOS_MODALIDADE.csv            8.8MB 合同类型代码表
│
├── programas_de_computador/        ← 计算机程序（9.8万+登记）
│   ├── PROG_COMP_DADOS_BASICOS.csv         7.2MB 基础信息
│   ├── PROG_COMP_AUTORES.csv              22.7MB 作者
│   ├── PROG_COMP_TITULAR.csv              18.7MB 权利人
│   ├── PROG_COMP_DESPACHOS.csv            80.6MB 批文
│   ├── PROG_COMP_CAMPOS.csv               134MB  技术领域
│   ├── PROG_COMP_LINGUAGEM.csv             6.2MB 编程语言
│   └── PROG_COMP_TIPO.csv                 26.8MB 类型
│
└── desenhos_industriais/          ← 工业设计（20.3万+）
    ├── DI_DEPOSITANTES.csv                 90MB  申请人
    ├── DI_DESPACHOS.csv                   118MB  批文
    └── ... 另有8个CSV + 3个图片ZIP

AI 查询逻辑

查询某CNPJ的商标:
  1. detect_cache() → 有缓存？→ 本地读MARCAS_DEPOSITANTES.csv
  2. 无缓存 → 后台curl下载835MB → 缓存 → 查询
  3. 返回: 商标名称、申请号、Nice分类、状态、持有人

查询某公司的专利:
  1. detect_cache() → 有缓存？→ 本地读PATENTES_DEPOSITANTES.csv
  2. 无缓存 → 后台curl下载138MB → 缓存 → 查询
  3. 返回: 专利标题、IPC分类、发明人、法律状态

查询技术合同:
  1. 按当事人名/CNPJ查 CONTRATOS_CESSIONARIOS.csv
  2. JOIN CONTRATOS_MODALIDADE.csv 获取合同类型描述
  3. JOIN CONTRATOS_DESPACHO.csv 获取法律状态

在线备选方案（无本地数据时）

• INPI pePI (busca.inpi.gov.br)：无官方API，仅网页搜索，结果最多100条
• INPI Portal de Servicos：Beta阶段，API v1.0.2计划中
• 数据滞后：INPI 数据通常滞后 1~4 周（与 RPI 公告周期同步）

数据时效性参考（了解即可，在线查询无此问题）

RFB 数据特征（仅对需要离线高级模式的用户有用）：

附录：本地缓存查询（AI 自动执行，用户无感）

AI 自动检测 ~/.workbuddy/brazil-data/ 下的缓存数据，按需从官方源下载。 用户全程无需手动操作。首次查询自动下载，后续毫秒级响应。

缓存数据路径与官方源对照

缓存位置                    数据内容              来源官方源
─────────────────────────────────────────────────────────────────
~/.workbuddy/brazil-data/
├── cnpj-data/parquet/      RFB CNPJ 企业登记     RFB WebDAV
│   ├── empresas/0.parquet ~ 9.parquet  (1.4GB)
│   ├── estabelecimentos/~ (5.2GB)
│   ├── socios/0.parquet ~ 9.parquet    (677MB)
│   ├── simples/0.parquet               (297MB)
│   └── cnaes/、municipios/、naturezas/...
│
└── inpi/                    INPI 知识产权        dadosabertos.inpi.gov.br
    ├── marcas/MARCAS_DEPOSITANTES.csv          (835MB)
    ├── patentes/PATENTES_DEPOSITANTES.csv       (138MB)  ← 新增！
    ├── contratos/CONTRATOS_DESPACHO.csv         (302MB)
    ├── contratos/CONTRATOS_CEDENTE.csv           (36MB)
    ├── contratos/CONTRATOS_CESSIONARIOS.csv      (97MB)
    └── programas_computador/                    (297MB)

数据格式说明

AI 自动查询逻辑（缓存优先 + 按需下载）

首次查询流程（缓存不存在时）

import os, subprocess, pyarrow.parquet as pq, pandas as pd

BRAZIL_CACHE = os.path.expanduser("~/.workbuddy/brazil-data")
CNPJ_DATA = f"{BRAZIL_CACHE}/cnpj-data/parquet"
INPI_DATA = f"{BRAZIL_CACHE}/inpi"
RFB_SOURCE = "https://arquivos.receitafederal.gov.br/public.php/dav/files/YggdBLfdninEJX9"

def query_company(cnpj_or_name):
    """查询企业信息：缓存优先 → 按需下载 → 在线兜底"""
    is_cnpj = len(cnpj_or_name.replace(".","").replace("/","").replace("-","")) == 14

    if is_cnpj:
        cnpj_clean = cnpj_or_name.replace(".","").replace("/","").replace("-","")
        cnpj_8 = cnpj_clean[:8]
        shard = int(cnpj_8) % 10

        # 检查缓存
        empresas_parquet = f"{CNPJ_DATA}/empresas/{shard}.parquet"
        if not os.path.exists(empresas_parquet):
            # 无缓存 → 按需下载
            month = "2026-06"  # AI动态获取最新月份
            for f in [f"Empresas{shard}.zip"]:
                dl_url = f"{RFB_SOURCE}/{month}/{f}"
                dl_target = f"{CNPJ_DATA}/{f}"
                os.makedirs(f"{CNPJ_DATA}/empresas", exist_ok=True)
                # AI执行: curl -L -o {dl_target} {dl_url}
                # AI执行: unzip -o {dl_target} -d {CNPJ_DATA}
                pass

        # 本地查询
        df = pq.read_table(empresas_parquet).to_pandas()
        result = df[df['cnpj_8'] == cnpj_8]

        # 补充最新状态：WebSearch
        # ...

        return result
    else:
        # 按公司名搜索 → 扫描全部10个分片（缓存已存在时）
        for i in range(10):
            p = f"{CNPJ_DATA}/empresas/{i}.parquet"
            if os.path.exists(p):
                df = pq.read_table(p).to_pandas()
                found = df[df['razao_social'].str.contains(cnpj_or_name, case=False, na=False)]
                if len(found) > 0: return found
        # 全无缓存 → WebSearch在线查
        return websearch_company(cnpj_or_name)

按 CNPJ 查询企业信息（缓存命中时 — 最快路径）

CACHE = os.path.expanduser("~/.workbuddy/brazil-data/cnpj-data/parquet")
cnpj_8 = target_cnpj[:8]
shard = int(cnpj_8) % 10  # 计算分片号，只查1个分片

# 查询 empresas 表 → 企业基本信息
df = pq.read_table(f"{CACHE}/empresas/{shard}.parquet").to_pandas()
row = df[df['cnpj_8'] == cnpj_8]

# 查询 estabelecimentos 表 → 分支机构详情
est = pq.read_table(f"{CACHE}/estabelecimentos/{shard}.parquet").to_pandas()
est_row = est[est['cnpj_8'] == cnpj_8]

# 查询 socios 表 → 股东结构
soc = pq.read_table(f"{CACHE}/socios/{shard}.parquet").to_pandas()
soc_row = soc[soc['cnpj_8'] == cnpj_8]

按公司名搜索（缓存命中时）

CACHE = os.path.expanduser("~/.workbuddy/brazil-data/cnpj-data/parquet")
target = "公司名关键词"
for i in range(10):
    p = f"{CACHE}/empresas/{i}.parquet"
    if os.path.exists(p):
        df = pq.read_table(p).to_pandas()
        found = df[df['razao_social'].str.contains(target, case=False, na=False)]
        if len(found) > 0: break

查 INPI 商标（缓存优先 → 按需下载）

CACHE = os.path.expanduser("~/.workbuddy/brazil-data/inpi")
marcas_file = f"{CACHE}/marcas/MARCAS_DEPOSITANTES.csv"

if not os.path.exists(marcas_file):
    # 按需下载 (AI自动执行)
    # curl -L -o {marcas_file} https://dadosabertos.inpi.gov.br/download/marcas/MARCAS_DEPOSITANTES.csv
    pass

# 按 CNPJ 查商标申请/持有人
marcas = pd.read_csv(marcas_file, dtype=str, low_memory=False)
result = marcas[marcas['cnpj_cpf_titular'].str.contains(cnpj_8, na=False)]

查 INPI 专利（缓存优先 → 按需下载）

CACHE = os.path.expanduser("~/.workbuddy/brazil-data/inpi")
patentes_file = f"{CACHE}/patentes/PATENTES_DEPOSITANTES.csv"

if not os.path.exists(patentes_file):
    # curl -L -o {patentes_file} https://dadosabertos.inpi.gov.br/download/patentes/PATENTES_DEPOSITANTES.csv
    pass

patentes = pd.read_csv(patentes_file, dtype=str, low_memory=False)
result = patentes[patentes['cnpj_cpf_depositante'].str.contains(cnpj_8, na=False)]

查 INPI 技术合同（缓存优先 → 按需下载）

CACHE = os.path.expanduser("~/.workbuddy/brazil-data/inpi")
contratos_file = f"{CACHE}/contratos/CONTRATOS_DESPACHO.csv"

if not os.path.exists(contratos_file):
    # curl -L -o {contratos_file} https://dadosabertos.inpi.gov.br/download/contratos/CONTRATOS_DESPACHO.csv
    pass

contratos = pd.read_csv(contratos_file, dtype=str, low_memory=False)
result = contratos[contratos['nm_parte'].str.contains("关键词", case=False, na=False)]

参考资料

本技能还引用以下参考资源：

Doing Business in Brazil (7th Edition) — Britcham, 2020

• 位置: references/Doing_Business_in_Brazil_7th.txt
• 内容: 536页PDF，含992,850字符（约165,000单词），涵盖巴西商业环境全貌
• 章节: 巴西概况 → 政治制度 → 公司法 → 税务 → 劳工 → 移民 → 环境 → 知识产权 → 竞争法 → 公共事业 → 国际贸易 → 争议解决
• 用途: 回答关于巴西商业环境、法律法规、投资政策、税务制度等宏观问题时优先参考此文档

Drivers of Trust in Public Institutions in Brazil — OECD, 2023

• 位置: references/OECD_Trust_in_Public_Institutions_Brazil_2023.txt
• 内容: OECD 经合组织报告，161页，含459,521字符（约76,000单词）
• 核心主题: 巴西公民对公共机构的信任度驱动因素分析，涵盖政府透明度、反腐败、公共服务质量、司法体系信任度等
• 用途: 回答巴西政治环境、政府治理、反腐败、司法效能等政策层面问题时优先参考此文档

World Bank Subnational Doing Business in Brazil 2021

• 位置: references/World_Bank_Subnational_Brazil_2021.txt
• 内容: 354页报告，1,254,939字符（约209,000单词），覆盖巴西27城市 vs 190经济体的营商环境排名
• 核心主题: 各州/市的开办企业、施工许可、产权登记、合同执行、纳税等指标排名
• 用途: 按城市提供企业开办、合同执行、纳税便利度等微观营商环境数据

World Bank Reports

• 位置1: references/World_Bank_1.txt（125页，327,563字符）
• 位置2: references/World_Bank_2.txt（70页，176,751字符）
• 核心主题: 世界银行涉及巴西的营商环境、经济发展、贸易政策等相关报告
• 用途: 补充宏观经济发展趋势、跨境投资环境等主题的参考

INPI 开放数据与CNPJ数据库分析报告

• 位置: references/INPI_API_CNPJ_database_analysis.md
• 内容: INPI 各API/数据库的技术分析，涵盖专利、商标、工业设计、计算机程序、技术转让合同五大类
• 关键发现: 华为是巴西第二大专利申请人(18,753件)，中国排第7(88,847件)；旧版pePI系统与新Portal开发路线图

在线 API 资源（实时查询，无需下载）

以下 API 可在回答问题时实时调用获取数据：

世界银行国家/指标 API

• 端点: https://api.worldbank.org/v2/
• 认证: 无需 Key，完全开放
• 巴西基础信息: GET /v2/country/br?format=json
• 经济指标: GET /v2/country/BR/indicator/{code}?format=json
• 可用指标分类: 私营部门(营商环境)、贸易、金融、税收、基础设施、劳动力等数百种
• 用途: 查询巴西历年GDP、营商便利度排名、各经济指标等实时数据

与企业最相关的巴西指标代码（✅ 已验证可用）：

⚠️ 原表中营商便利度、企业开办成本、总税率、进出口耗时、信贷便利度、基尼系数等指标已被世界银行归档下架（2021年Doing Business项目终止），不可用。如需此类数据，请使用 WebSearch 查询最新报告。

Datajud 巴西司法诉讼 API（公共 Key，零注册）

关键发现：CNJ 在官方 Wiki 页面提供公共 API Key，无需任何注册即可使用。 AI 自动从 Wiki 页面实时抓取当前 Key（因为 Key 会被轮换），用户完全无感。

端点: https://api-publica.datajud.cnj.jus.br/api_publica_{sigla_tribunal}/_search 方法: POST（Elasticsearch Query DSL，请求体为JSON） 请求头: Authorization: APIKey {当前公共Key} | Content-Type: application/json Key 来源（AI自动抓取）:

# AI 每次调用前自动执行（无需用户操作）
import re, urllib.request
WIKI_URL = "https://datajud-wiki.cnj.jus.br/api-publica/acesso/"
html = urllib.request.urlopen(WIKI_URL, timeout=20).read().decode('utf-8')
DATAJUD_KEY = re.search(r'Authorization:\s*APIKey\s*<strong>([^<]+)</strong>', html).group(1).strip()
# 测试: curl -X POST {endpoint} -H "Authorization: APIKey {DATAJUD_KEY}" -H "Content-Type: application/json" -d '...'

⚠️ 重要限制：Datajud 公开 API 因 LGPD（巴西数据保护法）限制，不支持按 CNPJ 搜索。只能按当事人名称（partes.nome）或案件号（numeroProcesso）搜索。AI 查询企业诉讼时应先用公司名搜索，再人工确认。

法院索引（sigla_tribunal）对照表：

Elasticsearch 查询参数（主要可过滤字段）：

查询示例（Python）：

import re, urllib.request, json

# Step 1: 自动获取当前公共 Key
WIKI_URL = "https://datajud-wiki.cnj.jus.br/api-publica/acesso/"
html = urllib.request.urlopen(WIKI_URL, timeout=20).read().decode('utf-8')
API_KEY = re.search(r'Authorization:\s*APIKey\s*<strong>([^<]+)</strong>', html).group(1).strip()

headers = {"Authorization": f"APIKey {API_KEY}", "Content-Type": "application/json"}

# Step 2: 按公司名查诉讼（替代CNPJ搜索）
url = "https://api-publica.datajud.cnj.jus.br/api_publica_tjsp/_search"
query = {
    "size": 20,
    "query": {"match": {"partes.nome": "Petrobras"}},
    "sort": [{"dataAjuizamento": {"order": "desc"}}]
}

resp = requests.post(url, headers=headers, json=query)
for hit in resp.json().get("hits", {}).get("hits", []):
    p = hit["_source"]
    print(p.get('numeroProcesso'), p.get('classe',{}).get('nome'), p.get('dataAjuizamento'))

关键返回字段：

• numeroProcesso - 统一案件号
• classe.codigo / classe.nome - 案件类型
• dataAjuizamento - 立案日期
• grau - 审级
• nivelSigilo - 保密级别
• orgaoJulgador.codigo / .nome / .municipio - 管辖机构
• assuntos[].codigo / .nome / .principal - 案件主题
• movimentos[].codigo / .nome / .dataHora - 案件动态

多法院并行查询策略（企业诉讼排查用）：

import asyncio

TRIBUNAIS = ['tjsp','tjrj','tjmg','trf1','trf3','trt2','trt15','stj','tst']
async def search_all(company_name):
    async with aiohttp.ClientSession(headers=headers) as session:
        tasks = []
        for t in TRIBUNAIS:
            url = f"https://api-publica.datajud.cnj.jus.br/api_publica_{t}/_search"
            tasks.append(session.post(url, json={
                "size": 10,
                "query": {"match": {"partes.nome": company_name}}
            }))
        results = await asyncio.gather(*tasks)
    return [await r.json() for r in results]

注意事项：

• 公共 Key 会被 CNJ 不定期轮换，AI 自动从 Wiki 页面实时抓取最新 Key
• 部分法院数据质量差异（数据延迟不定）
• 保密案件（nivelSigilo>1）返回数据有限
• 建议同时搜索多个法院覆盖全国
• 高峰期 API 可能不稳定，建议实现指数退避重试

BCB 巴西中央银行开放数据

• 门户: https://dadosabertos.bcb.gov.br/
• API类型: CKAN API (/api/3/action/) + OData
• 用途: 查询巴西金融/经济实时数据

与企业最相关的BCB数据集：

BCB SGS 时间序列 API（实时数据查询）

• 端点: https://api.bcb.gov.br/dados/serie/bcdata.sgs.{codigo}/dados/ultimos/{n}?formato=json
• 调用说明: {codigo} = SGS 序列号, {n} = 最近N条数据(最大20), 支持日期范围查询: ?formato=json&dataInicial=dd/mm/yyyy
• 注意: ultimos/ 参数最大值是20，超过会报错

已验证的巴西核心经济指标 SGS 代码：

过去一年 USD/BRL 汇率波动分析（2025.06.18 - 2026.06.17）：

• 最高：5.6028（2025/07/30）
• 最低：4.8967（2026/05/11）
• 波动率：(5.6028 - 4.8967) / 4.8967 ≈ 14.42%
• 当前(2026/06/17)相比去年高点已回落约 9.6%

实时数据调用示例：

• GET /bcdata.sgs.10813/dados/ultimos/5?formato=json → 美元汇率最新5天
• GET /bcdata.sgs.11/dados/ultimos/3?formato=json → Selic利率
• GET /bcdata.sgs.433/dados/ultimos/12?formato=json → 最近12个月IPCA通胀
• GET /bcdata.sgs.13621/dados/ultimos/5?formato=json → 国际储备

定价报价查询工作流

当专家需要进行批量产品的定价报价分析时（用户询问「这个价格能不能做」等场景），工作流如下：

1. 获取产品清单：从用户输入解析产品名称、数量、单价（CNY）
2. 实时查询BCB API序列：
   • 10813 最新5天 → 获取当前 USD/BRL 汇率
   • 10813 日期范围查询 → 获取过去1年数据，计算波动率
   • 11 最新 → 获取 Selic 利率
   • 433 最近13个月 → 获取年度 IPCA 通胀率
3. 换算逻辑：CNY → USD → BRL（或直接 CNY/BRL 交叉汇率）
4. 美元结算建议：
   • BRL贬值 >8%(3月内) + 通胀 >6% + 波动 >10% → ✅ 强烈建议美元
   • 波动 5-10% + 通胀 4-6% → ⚠️ 混合结算
   • BRL稳定/升值 + 通胀 <4% + 波动 <5% → ❌ 可用BRL
5. 输出：报价评估表 + 宏观经济参考表

信用风险评估工作流

当报价评估表生成后，专家必须自动执行信用风险评估。评估自动识别目标企业/行业/城市，依次调用以下数据源：

数据源调用顺序：

1. 如已有 CNPJ → 查询 RFB 工商状态（ativa/suspensa/inapta/baixada/nula）和税务制度
2. 如已有 CNPJ → 查询 Datajud API 检查司法诉讼记录
3. 如已有 CNPJ → 查询 INPI 数据库确定知识产权持有情况
4. 如仅有行业 → 引用 Doing Business 7th 及 OECD 报告的行业风险分析
5. 如仅有城市 → 引用 World Bank Subnational Brazil 2021 的城市排名

举措建议矩阵（综合风险等级决定）：

• ✅ 低风险：30%+70%见提单副本，标准合同
• ⚠️ 中风险：50%+50%发货前付清，巴西律师审合同
• ❌ 高风险：100% T/T预付或保兑不可撤销L/C，中国信保承保
• 🔴 极高风险：100%全款到账后发货，建议放弃该客户

关键参考文本中的案例方向：

• Doing Business 7th: 司法效率(2-4年)、Lava Jato反腐败、破产受偿顺序、CARF税务争议
• OECD 2023: 制度信任与公共服务效率
• World Bank 2021: 城市间营商环境差异（São Paulo vs Salvador）

⚠️ 以上内容由 AI 基于公开信息整理生成，仅供参考，不构成任何投资建议或个股推荐。投资有风险，决策需谨慎。

出海法律提醒工作流

在报价评估 + 信用风险评估完成之后，检测到用户有向巴西出口意图时，自动输出本模块。

触发信号：用户提及「巴西客户」「出口巴西」「发往巴西」「巴西市场」或询问具体产品能否进入巴西。

产品-监管机构映射：

• 食品/农产品 → MAPA + ANVISA
• 药品/医疗器械/化妆品 → ANVISA
• 电子产品/家电/IT → INMETRO + ANATEL
• 化工/农药/肥料 → IBAMA + ANVISA + MAPA
• 汽车/零部件/机械 → INMETRO + CONTRAN + IBAMA
• 玩具 → INMETRO（强制认证）
• 建材/钢材 → INMETRO + ABNT
• 医疗器械 → ANVISA（注册+GMP）+ INMETRO
• 能源/电池 → INMETRO + ANEEL + IBAMA

主要监管要求（引自语料库 Doing Business 7th）：

• ANVISA：药品注册、医疗器械分类注册+GMP、化妆品登记、食品登记
• INMETRO：强制认证（产品测试+工厂检查+市场监督）、葡萄牙语标签、ILAC互认
• MAPA：动物源产品健康证书、植物检疫证书、农药登记、双边协议认可名单
• IBAMA：化学品环境许可证、农药双重登记（与MAPA）
• ANATEL：电信设备三类认证、认可实验室测试报告

关键要求：

• 所有进口消费品需要葡萄牙语标签（产品名+CNPJ+原产地+成分+净含量+使用说明+安全警告）
• 部分品类必须通过对应机构认证/注册后方可进口
• 农药登记周期 2-5 年（全球最严格之一）
• 高风险医疗器械需境外工厂 GMP 现场检查

在线搜索确认规则：

• 每次输出后必须搜索最新法规变更
• 搜索中巴双边协议更新
• 搜索近期巴西召回/禁入事件
• 标记来源：语料库内容 vs 🔍在线确认内容

法律案例查询工作流

用户提出巴西法律相关问题（合同纠纷、劳工诉讼、税务争议、知识产权侵权、环保处罚等）时，自动激活。

语料库案例检索优先级：

1. Doing Business 7th Ch.22 Dispute Resolution: 法院管辖、诉讼程序、仲裁制度、证据规则
2. Doing Business 7th Ch.9 Labour Law: 劳工法院执行、解雇补偿、集体谈判
3. Doing Business 7th Ch.6 Tax Law: CARF税务法庭、税务争议
4. Doing Business 7th Ch.12 Environmental Law: IBAMA处罚、生物多样性法
5. Doing Business 7th Ch.10 IP: 商标/专利侵权、INPI无效程序
6. OECD Trust Report: 司法效率与信任度
7. Lava Jato反腐败 (Doing Business 7th + OECD): 腐败定罪、合规要求

用户行为提醒规则：合同仲裁条款、书面合同要求、税务合规、认证前置、劳工法执行、环境许可等针对性提醒。

本地化经营指南工作流

用户表现出去巴西开分公司/子公司/办事处的意向后自动激活。

设立形式对比：分公司(Branch, 需总统法令) / LTDA有限责任公司(推荐) / S.A.股份有限公司 / 代表处

关键城市信用数据（参考 WB Subnational Brazil 2021）：

• ✅ 低风险城市：São Paulo, Belo Horizonte, Curitiba, Porto Alegre, Florianópolis
• ⚠️ 中风险城市：Rio de Janeiro, Brasília, Manaus(自贸区)
• ❌ 较高风险城市：Fortaleza, Salvador, Recife（执行合同3-4年）

10步注册清单：文件准备(海牙认证) → 城市选择 → 律师聘请 → Junta Comercial注册 → CNPJ → 经营许可 → 银行开户 → 会计师 → 劳工登记

费用参考：圣保罗设立LTDA初始成本BRL 5,000-15,000

⚠️ 以上内容由 AI 基于公开信息整理生成，仅供参考，不构成任何投资建议或个股推荐。投资有风险，决策需谨慎。