85 lines
2.5 KiB
Python
85 lines
2.5 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
BIS 实体清单采集与匹配模块
|
||
支持企业名模糊匹配 + 别名映射 + 供应链上游穿透
|
||
"""
|
||
import json
|
||
import logging
|
||
from pathlib import Path
|
||
from typing import Optional
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
DATA_DIR = Path(__file__).parent.parent / "data"
|
||
|
||
|
||
def _load_entity_list() -> list:
|
||
"""加载实体清单数据"""
|
||
filepath = DATA_DIR / "entity_list.json"
|
||
with open(filepath, "r", encoding="utf-8") as f:
|
||
return json.load(f)
|
||
|
||
|
||
def check_entity_list(company_name: str) -> dict:
|
||
"""
|
||
检查企业是否在 BIS 实体清单中
|
||
支持模糊匹配和别名匹配
|
||
"""
|
||
entities = _load_entity_list()
|
||
result = {
|
||
"is_sanctioned": False,
|
||
"match_type": None,
|
||
"entity_detail": None,
|
||
"supply_chain_risk": [], # 供应链上游被制裁的情况
|
||
}
|
||
|
||
for entity in entities:
|
||
# 精确匹配
|
||
if company_name in entity["entity_name"]:
|
||
result["is_sanctioned"] = True
|
||
result["match_type"] = "直接命中"
|
||
result["entity_detail"] = entity
|
||
return result
|
||
|
||
# 别名匹配
|
||
for alias in entity.get("aliases", []):
|
||
if company_name in alias or alias in company_name:
|
||
result["is_sanctioned"] = True
|
||
result["match_type"] = "别名命中"
|
||
result["entity_detail"] = entity
|
||
return result
|
||
|
||
return result
|
||
|
||
|
||
def check_supply_chain_sanctions(company_name: str, suppliers: list) -> list:
|
||
"""
|
||
检查企业供应链上游是否有被制裁的实体
|
||
返回受制裁的供应商列表
|
||
"""
|
||
sanctioned_suppliers = []
|
||
entities = _load_entity_list()
|
||
|
||
for supplier in suppliers:
|
||
# 清洗供应商名称（去掉括号中的说明文字）
|
||
clean_name = supplier.split("（")[0].split("(")[0].strip()
|
||
|
||
for entity in entities:
|
||
all_names = [entity["entity_name"]] + entity.get("aliases", [])
|
||
for name in all_names:
|
||
if clean_name in name or name in clean_name:
|
||
sanctioned_suppliers.append({
|
||
"supplier": supplier,
|
||
"matched_entity": entity["entity_name"],
|
||
"restrictions": entity["restrictions"],
|
||
"date_added": entity["date_added"],
|
||
})
|
||
break
|
||
|
||
return sanctioned_suppliers
|
||
|
||
|
||
def get_all_sanctioned_entities() -> list:
|
||
"""获取所有被制裁实体列表"""
|
||
return _load_entity_list()
|