微信扫一扫企业触达关系查询
触发条件
用户请求查找两个实体（企业/个人）之间的关联关系、触达关系、关系路径、股权关联、法人关联、上下游关联、实际控制人关联等需求时触发。
使用前提
本技能依赖企智多平台的专有数据接口，使用前需由用户自行提供 API KEY（脚本不内置任何密钥）：
- 获取 API KEY：在企智多平台（www.qizhiduo.com）自行获取；
- 配置 API KEY：运行数据采集脚本时通过
--api-key参数传入，或配置环境变量QZD_API_KEY；所有接口请求在请求头携带Authorization: Bearer <API KEY>。
输出
当用户触发本技能但尚未提供 API KEY 时，请向用户输出以下内容：
「企业触达关系查询」技能需要调用企智多平台的专有数据接口，请先在企智多平台（www.qizhiduo.com）获取 API KEY，并提供后再继续查询。
禁止行为：
- 不得尝试自行编造企业/个人之间的关联关系或关系路径
- 不得使用其他数据源替代本技能的数据接口进行近似分析
- 仅引导用户提供合法 API KEY
安全与隐私说明
- 单一域名访问：脚本仅请求
https://www.qizhiduo.com一个域名（即数据来源），不访问任何其他外部地址； - 凭据安全：API KEY 由用户在运行时提供，仅用于上述域名的请求鉴权，脚本不硬编码、不持久化密钥，不向任何其他地址发送密钥；
- 无隐私采集：不读取、不上传用户本地文件、环境变量或其他隐私信息，产出仅为接口公开返回的企业/人员关系数据。
执行方式（强制）
本技能的全部数据统一由下方「数据采集脚本」章节中的 Python 脚本获取。执行本技能必须严格按以下步骤操作：必须实际运行该脚本获取所有数据，再基于脚本产出的数据严格按照「Output」规范与模板生成结果。
Workflow
请严格按照以下步骤执行，每一步都基于上一步的结果：
1. 获取 API KEY
使用本技能前，必须询问用户提供 API KEY。若用户尚未提供或未配置 API KEY，按照上文"输出"章节的内容引导用户获取。
2. 设置请求头与请求格式
得到 API KEY 后，后续所有接口请求统一携带以下请求头：
Authorization: Bearer <API KEY>User-Agent: PostmanRuntime/7.37.0（必须设置：网关会拦截默认 curl 的 User-Agent 并返回 "Page Not Found"，使用该值可正常访问）
表单提交说明（重要）： 查企业、信用代码加密、查个人 PID 三个接口采用表单提交（application/x-www-form-urlencoded），不是 JSON body：把 params 作为表单字段，其值为一段 JSON 字符串（文档中标注"表单参数 params 为 JSON 字符串"的接口均按此格式）。保存记录（toSave）与关系路径查询（pathQueryNew）两个接口使用 JSON body。
3. 解析两个实体（起始节点与结束节点）
根据用户问题识别两个需要查关系的实体：
- 实体可能都是企业（如"查找京信科技和中山电信的关系"）→ 两个节点类型均为 0（企业）；
- 实体可能一个是企业一个是个人（如"查找京信科技和黎刚的关系"）→ 企业节点类型 0，个人节点类型 1；
- 若用户只给两个名称且无法确认类型，按"先企业、后个人"判断，必要时结合脚本输出结果与用户确认。
每个节点需要确定：节点名称（start_name/end_name）、节点类型（start_type/end_type，0=企业，1=个人）。
4. 解析节点信息
（1）企业节点：调用查企业接口检索并确认企业，提取企业名称、统一社会信用代码，再调用信用代码加密接口获取加密后的 qydm。企业名称存在多个候选时优先选择名称精确匹配的记录，否则取第一条；用户已给出统一社会信用代码时直接用 --start-tyshxydm / --end-tyshxydm 指定，跳过检索。
（2）个人节点：调用查个人 PID 接口，传入个人姓名（可同时传入其所在企业的统一社会信用代码以精确定位 PID），返回 PID 作为个人节点的 qydm。
5. 保存查询记录
调用 jxZgxRecord/toSave 保存一条 one-to-one 查询记录，提交起始/结束节点的 qydm、名称、法人/法人ID、简称、节点类型组成的 jsonData 以及关系类型列表，返回 recordId。
6. 关系路径查询
调用 jxZgxResult/pathQueryNew，提交 graphAlias、起始/结束节点信息（qydm、qymc、tyshxydm）、节点类型、最大深度 4、全路径方向 BOTH、关系类型列表以及上一步的 recordId，获取两个实体之间的全部关系路径。
7. 生成结果
按下方"Output"章节的规范与模板，结合脚本产出的关系路径数据生成最终结果。
数据来源清单（仅使用以下接口真实返回的数据）：
POST /api/api-zsai-query/zsai/user/query_es_dzxxQycmdGjcx 查企业（表单 params，keyword_flag=MH）
POST /api/api-form/form/core/formCustomQuery/queryForJson_qyhxQydm 信用代码加密（表单 params，dm=明文信用代码）
POST /api/api-form/form/core/formCustomQuery/queryForJson_ai_jxZtZdryDrFddbr 查个人PID（表单 params，pname[+tyshxydm]）
POST /api/api-dzxxfwpt/dzxxfwpt/jxZgxRecord/toSave 保存关系查询记录（JSON body）
POST /api/api-dzxxfwpt/dzxxfwpt/jxZgxResult/pathQueryNew 关系路径查询（JSON body）
JSON 关键字段对照：search_raw=查企业原始返回、enc_raw=信用代码加密原始返回、person_raw=查个人PID原始返回、to_save_raw=保存记录原始返回、relation_raw=关系路径查询原始返回、start_node/end_node=解析后的节点信息（qymc/tyshxydm/qydm）、record_id=查询记录ID（用于生成关系图谱链接）。
禁止行为：严禁编造企业/个人之间的关联关系、路径节点或关系类型；严禁绕过脚本直接调接口或修改脚本内接口地址、参数、请求头；严禁用其他数据源近似替代；数据缺失必须自然改写并标注"数据暂缺"，不得机械替换占位符或删除段落。
数据采集脚本（fetch_relation_data.py）
"""
企业触达关系查询数据采集脚本。
用法示例：
python fetch_relation_data.py "京信科技" "黎刚" --api-key "<API KEY>"
python fetch_relation_data.py "京信科技" "中山电信" --start-type 0 --end-type 0 --api-key "<API KEY>"
# 已知统一社会信用代码时可直接指定，跳过企业检索
python fetch_relation_data.py "京信科技" "黎刚" --end-tyshxydm 914420006682263412 --api-key "<API KEY>"
输出：relation_data.json（含解析后的节点信息、record_id、关系路径原始返回与结构化路径）
"""
import argparse
import json
import os
import sys
import urllib.parse
import urllib.request
from datetime import datetime
BASE = "https://www.qizhiduo.com"
TIMEOUT = 30
HEADERS_BASE = {
# 必须设置：网关会拦截默认 UA 并返回 Page Not Found
"User-Agent": "PostmanRuntime/7.37.0",
"Origin": BASE,
"Referer": BASE + "/",
"Accept": "application/json, text/plain, */*",
}
if hasattr(sys.stdout, "reconfigure"):
try:
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
except Exception:
pass
# 关系类型（与查询接口保持一致）
PATHS = "分支机构,企业股东,企业法人,企业董监高,最终受益人,客户,实际控制人,供应商,招标中标"
PATHS_LIST = PATHS.split(",")
# ----------------------------------------------------------------------
# 基础请求与解析工具
# ----------------------------------------------------------------------
def _headers(api_key, content_type=None):
h = dict(HEADERS_BASE)
h["Authorization"] = "Bearer " + api_key
if content_type:
h["Content-Type"] = content_type
return h
def _request(method, url, api_key, form_params=None, json_body=None):
"""统一请求入口。
form_params: 表单提交，字段 params 的值为 JSON 字符串；
json_body: JSON 请求体。
失败（含超时）返回 {"__error__": ...}，不重试。"""
try:
if form_params is not None:
body = urllib.parse.urlencode(
{"params": json.dumps(form_params, ensure_ascii=False)}
).encode("utf-8")
req = urllib.request.Request(
url, data=body, method="POST",
headers=_headers(api_key, "application/x-www-form-urlencoded"))
elif json_body is not None:
body = json.dumps(json_body, ensure_ascii=False).encode("utf-8")
req = urllib.request.Request(
url, data=body, method=method,
headers=_headers(api_key, "application/json"))
else:
req = urllib.request.Request(url, method=method, headers=_headers(api_key))
with urllib.request.urlopen(req, timeout=TIMEOUT) as resp:
text = resp.read().decode("utf-8", errors="replace")
try:
return json.loads(text)
except Exception:
return {"__raw_text__": text[:5000]}
except Exception as exc:
return {"__error__": "%s: %s" % (type(exc).__name__, exc)}
def is_error(resp):
return isinstance(resp, dict) and "__error__" in resp
def best_record_list(obj, keys):
"""在任意深度的 JSON 中，找到与 keys 最相关的记录列表。"""
cands = []
def walk(o):
if isinstance(o, dict):
for v in o.values():
walk(v)
elif isinstance(o, list):
dicts = [x for x in o if isinstance(x, dict)]
if dicts:
cands.append(dicts)
for x in o:
walk(x)
walk(obj)
if not cands:
return []
def score(lst):
return sum(sum(1 for k in keys if d.get(k) not in (None, "", [])) for d in lst)
return max(cands, key=score)
def first_value(obj, key):
"""深度遍历，返回 key 的第一个非空值。"""
found = []
def walk(o):
if found:
return
if isinstance(o, dict):
for k, v in o.items():
if k == key and v not in (None, "", []):
found.append(v)
return
walk(v)
elif isinstance(o, list):
for x in o:
walk(x)
if found:
return
walk(obj)
return found[0] if found else None
# ----------------------------------------------------------------------
# 各接口封装（与 Workflow 逐条对应）
# ----------------------------------------------------------------------
def search_company(api_key, keyword):
"""第4步(1)：检索企业（表单提交，必须带 keyword_flag=MH）。"""
url = BASE + "/api/api-zsai-query/zsai/user/query_es_dzxxQycmdGjcx"
params = {
"keyword": keyword,
"keyword_flag": "MH",
"keyword_type": "qymc,sb",
"jyzt": "1,5",
"pageIndex": 1,
"pageSize": 5,
}
return _request("POST", url, api_key, form_params=params)
def encrypt_qydm(api_key, tyshxydm):
"""第4步(2)：以明文信用代码换取加密后的 qydm。"""
url = BASE + "/api/api-form/form/core/formCustomQuery/queryForJson_qyhxQydm"
resp = _request("POST", url, api_key, form_params={"pageIndex": 1, "dm": tyshxydm})
if is_error(resp):
return None, resp
recs = best_record_list(resp, ["qydm"])
qydm = recs[0].get("qydm") if recs else None
if not qydm:
qydm = first_value(resp, "qydm")
return qydm, resp
def search_person(api_key, pname, tyshxydm=""):
"""第4步(3)：查个人 PID（表单提交）。tyshxydm 可选，传入后精确定位该企业下的个人。"""
url = BASE + "/api/api-form/form/core/formCustomQuery/queryForJson_ai_jxZtZdryDrFddbr"
params = {"pname": pname}
if tyshxydm:
params["tyshxydm"] = tyshxydm
return _request("POST", url, api_key, form_params=params)
def to_save_record(api_key, json_data, start_name, end_name, start_qydm, end_qydm):
"""第5步：保存 one-to-one 关系查询记录，返回 recordId。"""
url = BASE + "/api/api-dzxxfwpt/dzxxfwpt/jxZgxRecord/toSave"
body = {
"name": "%s、%s" % (start_name, end_name),
"unique": "%s,%s" % (start_qydm, end_qydm),
"type": "one-to-one",
"jsonData": json.dumps(json_data, ensure_ascii=False),
"paths": PATHS,
}
return _request("POST", url, api_key, json_body=body)
def path_query(api_key, start_node, end_node, start_name, end_name,
start_type, end_type, record_id):
"""第6步：关系路径查询，返回全部关系路径。"""
url = BASE + "/api/api-dzxxfwpt/dzxxfwpt/jxZgxResult/pathQueryNew"
params = {
"graphAlias": "hugegraph_zwzc20250821",
"startNodeValue": start_name,
"endNodeValue": end_name,
"startNodeValueList": [{
"qydm": start_node["qydm"],
"qymc": start_node["qymc"],
"tyshxydm": start_node.get("tyshxydm", ""),
}],
"endNodeValueList": [{
"qydm": end_node["qydm"],
"qymc": end_node["qymc"],
"tyshxydm": end_node.get("tyshxydm", ""),
}],
"maxDepth": 4,
"depthType": "allPaths",
"direction": "BOTH",
"startNodeType": start_type,
"endNodeType": end_type,
"paths": PATHS_LIST,
}
if record_id:
params["recordId"] = record_id
return _request("POST", url, api_key, json_body=params)
# ----------------------------------------------------------------------
# 节点解析
# ----------------------------------------------------------------------
def resolve_enterprise(api_key, name, tyshxydm_override=""):
"""解析企业节点：检索企业 → 加密 qydm。返回 (node, err)。"""
raw_search = None
if tyshxydm_override:
recs = []
raw_search = {"__override__": tyshxydm_override}
else:
raw_search = search_company(api_key, name)
if is_error(raw_search):
return None, raw_search
recs = best_record_list(raw_search, ["tyshxydm", "qymc"])
hit = None
if tyshxydm_override:
hit = {"qymc": name, "tyshxydm": tyshxydm_override}
elif recs:
# 名称精确匹配优先，否则取第一条
hit = next((c for c in recs
if str(c.get("qymc") or "").strip() == name), None) or recs[0]
if hit is None:
return None, {"__error__": "未检索到企业：%s" % name}
qymc = str(hit.get("qymc") or name).strip()
tyshxydm = str(hit.get("tyshxydm") or tyshxydm_override or "").strip()
if not tyshxydm:
return None, {"__error__": "企业缺少统一社会信用代码：%s" % qymc}
qydm, raw_enc = encrypt_qydm(api_key, tyshxydm)
if not qydm:
return None, {"__error__": "企业信用代码加密失败：%s" % qymc}
return {
"qymc": qymc,
"tyshxydm": tyshxydm,
"qydm": qydm,
"fddbr": str(hit.get("fddbr") or ""),
"fddbr_id": "",
"qyjc": str(hit.get("qyjc") or ""),
"node_type": 0,
"raw_search": raw_search,
"raw_enc": raw_enc,
}, None
def resolve_person(api_key, pname, tyshxydm=""):
"""解析个人节点：查 PID。返回 (node, err)。"""
raw = search_person(api_key, pname, tyshxydm)
if is_error(raw):
return None, raw
pid = None
for d in best_record_list(raw, ["PID", "pid"]):
pid = d.get("PID") or d.get("pid")
if pid:
break
if not pid:
pid = first_value(raw, "PID") or first_value(raw, "pid")
return {
"qymc": pname,
"tyshxydm": tyshxydm,
"qydm": str(pid) if pid else "",
"fddbr": "",
"fddbr_id": str(pid) if pid else "",
"qyjc": "",
"node_type": 1,
"raw_person": raw,
}, None
# ----------------------------------------------------------------------
# 主流程
# ----------------------------------------------------------------------
def write_out(result, path):
with open(path, "w", encoding="utf-8") as f:
json.dump(result, f, ensure_ascii=False, indent=2)
def main():
ap = argparse.ArgumentParser(description="企业触达关系查询数据采集脚本")
ap.add_argument("start_name", help="起始节点名称（企业名称或个人姓名）")
ap.add_argument("end_name", help="结束节点名称（企业名称或个人姓名）")
ap.add_argument("--api-key", default=os.environ.get("QZD_API_KEY", ""),
help="API KEY（也可用环境变量 QZD_API_KEY 提供）")
ap.add_argument("--start-type", type=int, default=0, help="起始节点类型：0企业 1个人（默认0）")
ap.add_argument("--end-type", type=int, default=0, help="结束节点类型：0企业 1个人（默认0）")
ap.add_argument("--start-tyshxydm", default="", help="起始企业统一社会信用代码（可选，跳过检索）")
ap.add_argument("--end-tyshxydm", default="", help="结束企业统一社会信用代码（可选，跳过检索）")
ap.add_argument("--out", default="relation_data.json", help="输出 JSON 路径")
args = ap.parse_args()
if not args.api_key.strip():
print("[终止] 未提供 API KEY。请先在企智多平台 "
"(www.qizhiduo.com) 获取 API KEY，并通过 --api-key 或环境变量 QZD_API_KEY 提供。")
sys.exit(2)
api_key = args.api_key.strip()
result = {
"meta": {
"start_name": args.start_name,
"end_name": args.end_name,
"start_type": args.start_type,
"end_type": args.end_type,
"generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
},
"start_node": None,
"end_node": None,
"record_id": "",
"data": {},
"status": {},
"errors": [],
}
def note(step, ok, msg=""):
result["status"][step] = {"ok": bool(ok), "msg": msg}
if not ok and msg:
result["errors"].append("%s: %s" % (step, msg))
print("[%s] %s %s" % ("OK " if ok else "跳过", step, msg))
# ---------- 第4步：解析起始节点 ----------
if args.start_type == 0:
start_node, err = resolve_enterprise(api_key, args.start_name, args.start_tyshxydm)
else:
start_node, err = resolve_person(api_key, args.start_name,
args.end_tyshxydm if args.end_type == 0 else "")
if err:
note("resolve_start", False, str(err.get("__error__")))
write_out(result, args.out)
sys.exit(1)
result["start_node"] = start_node
result["data"]["search_raw"] = start_node.get("raw_search")
result["data"]["enc_raw"] = start_node.get("raw_enc")
result["data"]["person_raw"] = start_node.get("raw_person")
note("resolve_start", True,
"起始节点 %s（qydm=%s）" % (start_node["qymc"], start_node["qydm"]))
# ---------- 第4步：解析结束节点 ----------
if args.end_type == 0:
end_node, err = resolve_enterprise(api_key, args.end_name, args.end_tyshxydm)
else:
end_node, err = resolve_person(api_key, args.end_name,
args.start_tyshxydm if args.start_type == 0 else "")
if err:
note("resolve_end", False, str(err.get("__error__")))
write_out(result, args.out)
sys.exit(1)
result["end_node"] = end_node
note("resolve_end", True,
"结束节点 %s（qydm=%s）" % (end_node["qymc"], end_node["qydm"]))
# ---------- 构建 jsonData（含企业名称高亮与法人信息） ----------
def build_node(node, ref_enterprise, display_name, is_person):
if is_person:
return {
"qydm": node["qydm"],
"tyshxydm": node.get("tyshxydm", ""),
"qymc": ref_enterprise.get("qymc", ""),
"qymc_html": ref_enterprise.get("qymc", ""),
"fddbr": node["qymc"],
"fddbr_id": node["qydm"],
"qyjc": "",
"nodeType": 1,
}
return {
"qydm": node["qydm"],
"tyshxydm": node.get("tyshxydm", ""),
"qymc": node["qymc"],
"qymc_html": node["qymc"],
"fddbr": node.get("fddbr", ""),
"fddbr_id": "",
"qyjc": node.get("qyjc", ""),
"nodeType": 0,
}
ref_enterprise = start_node if args.start_type == 0 else (
end_node if args.end_type == 0 else {"qymc": ""})
json_data = {
"startNode": build_node(start_node, ref_enterprise,
args.start_name, args.start_type == 1),
"endNode": build_node(end_node, ref_enterprise,
args.end_name, args.end_type == 1),
}
# ---------- 第5步：保存查询记录 ----------
to_save_raw = to_save_record(api_key, json_data, args.start_name, args.end_name,
start_node["qydm"], end_node["qydm"])
result["data"]["to_save_raw"] = to_save_raw
record_id = None
if is_error(to_save_raw):
note("to_save", False, str(to_save_raw.get("__error__")))
else:
record_id = first_value(to_save_raw, "data")
if isinstance(record_id, list):
record_id = record_id[0] if record_id else None
result["record_id"] = record_id
note("to_save", True, "保存记录成功 recordId=%s" % record_id)
# ---------- 第6步：关系路径查询 ----------
relation_raw = path_query(api_key, start_node, end_node,
args.start_name, args.end_name,
args.start_type, args.end_type, record_id)
result["data"]["relation_raw"] = relation_raw
if is_error(relation_raw):
note("path_query", False, str(relation_raw.get("__error__")))
else:
paths = best_record_list(relation_raw, ["path", "paths", "relation", "relations"])
total = first_value(relation_raw, "total")
if not paths and total is None:
note("path_query", False, "未找到两者之间的关系路径")
else:
result["data"]["paths"] = paths
note("path_query", True,
"找到 %s 条关系路径（record_id=%s）" % (len(paths), record_id))
write_out(result, args.out)
print("[完成] 结果已写入 %s" % args.out)
if __name__ == "__main__":
main()
Output
以下为输出规范。有数据的变量替换为实际值；关系路径数据缺失时，不得机械替换占位符，必须将所在句子改写为自然通顺的表述，不得删除该段落。
1. 查询说明
> 为用户查询「{{起始实体}}」与「{{结束实体}}」之间的关联关系路径，覆盖股权关系、法人关系、董监高关系、分支机构关系、最终受益人关系、客户关系、实际控制人关系、供应商关系、招标中标关系等。起始节点：{{起始节点名称}}（{{企业/个人}}）；结束节点：{{结束节点名称}}（{{企业/个人}}）。
2. 关系路径结果
输出关系路径概述（用一段话）： 说明是否找到两者之间的关联关系，共找到多少条关系路径，主要经由哪些关系类型（如股权、法人、董监高等）相连。
- 找到关系路径时：逐条输出路径。每条路径给出节点序列与相邻节点间的关系类型，例如： > 路径1：{{企业A}} —{{股权关系}}→ {{中间企业/个人}} —{{法人关系}}→ {{企业B}} > 路径2：{{企业A}} —{{董监高关系}}→ {{中间个人}} —{{分支机构关系}}→ {{企业B}} 路径较多时优先展示路径较短（节点更少）的路径，并控制输出条数（最多输出全部，数量过多时可先列出前若干条并说明总数）。
- 未找到关系路径时：输出"未查询到「{{起始实体}}」与「{{结束实体}}」之间的关联关系路径"，并说明可能原因（如两者无直接或间接关联），不编造任何关系。
3. 补充说明
> 以上关系路径基于企智多平台公开数据接口实时查询得出，关系类型包括：分支机构、企业股东、企业法人、企业董监高、最终受益人、客户、实际控制人、供应商、招标中标。数据以接口返回为准，如有遗漏请以最新查询结果为准。
数据缺失处理：
- 接口请求失败/超时 → 该接口数据按"数据暂缺"处理，其余可用数据正常输出；
- 未返回数据时标注"数据暂缺"或自然改写为"XX数据未查询到"。