Files
XH-202626/collectors/compliance_collector.py

88 lines
2.7 KiB
Python

# -*- coding: utf-8 -*-
"""
算法备案合规数据采集与查询模块
匹配企业是否已完成网信办算法备案
"""
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_algo_filings() -> list:
"""加载算法备案数据"""
filepath = DATA_DIR / "algo_filings.json"
with open(filepath, "r", encoding="utf-8") as f:
return json.load(f)
def check_algo_filing(company_name: str) -> dict:
"""
查询企业的算法备案状态
"""
filings = _load_algo_filings()
result = {
"has_filing": False,
"filings": [],
"needs_filing": False, # 是否需要备案但未备案
"risk_level": "低",
}
for filing in filings:
if company_name in filing["company"] or filing["company"] in company_name:
result["has_filing"] = True
result["filings"].append(filing)
return result
def assess_algo_compliance_risk(company_data: dict) -> dict:
"""
综合评估企业的算法合规风险
考虑因素:是否涉及 AI 业务、是否已备案、数据出境风险
"""
company_name = company_data.get("short_name", company_data.get("company_name", ""))
sector = company_data.get("sector", "")
compliance = company_data.get("compliance", {})
# 查询备案状态
filing_status = check_algo_filing(company_name)
# 判断是否需要备案
ai_related_sectors = ["AI", "软件", "互联网", "消费电子"]
needs_filing = sector in ai_related_sectors or "AI" in str(company_data.get("tech_route", {}))
# 综合评估
risk_level = "低"
risk_details = []
if needs_filing and not filing_status["has_filing"]:
algo_status = compliance.get("algo_filing_status", "")
if algo_status == "不适用":
risk_level = "低"
else:
risk_level = "高"
risk_details.append("涉及AI业务但未查到算法备案记录")
data_export_risk = compliance.get("data_export_risk", "低")
if data_export_risk == "高":
risk_level = "高"
risk_details.append("存在大量跨境数据传输,数据出境评估风险高")
elif data_export_risk == "中":
if risk_level != "高":
risk_level = "中"
risk_details.append("存在部分跨境数据传输,需关注数据出境合规")
return {
"company_name": company_name,
"needs_filing": needs_filing,
"filing_status": filing_status,
"data_export_risk": data_export_risk,
"overall_risk_level": risk_level,
"risk_details": risk_details,
}