影刀RPA Neo4j图数据库自动化知识图谱构建与查询作者林焱 | 适用人群影刀RPA新手 | 难度★★★★☆一、什么情况用这个做数据分析的兄弟经常需要搞清楚「A公司和B公司是什么关系」「这个人和哪些企业有关联」「这条供应链上有几个环节」。这种关系型查询用MySQL写SQL能把人写吐——三度以内的人脉关系SQL要JOIN三四次性能堪忧。图数据库Neo4j专门解决这种问题。但Neo4j的Cypher查询语言和传统SQL差别大很多人不熟悉。用影刀RPA定时从各种数据源采集关系数据 → 自动写入Neo4j → 定时查询生成关系图谱报告省去手动维护的麻烦。实际场景企业股权穿透分析、供应链上下游梳理、社交网络关系挖掘、知识图谱自动更新。这篇文章能解决的问题在影刀Python节点中连接Neo4j自动从Excel/网页采集关系数据并写入图数据库用Cypher查询做关联分析将查询结果导出为Excel报告二、怎么做2.1 环境准备影刀Python节点需要安装neo4j驱动拼多多店群自动化报活动上架影刀安装目录\python\python.exe-mpipinstallneo4j确保你有Neo4j数据库的访问地址默认bolt://localhost:7687和账号密码。2.2 影刀主流程┌──────────────────────────────────────────────┐ │ ① 【执行Python代码】 │ │ 连接Neo4j初始化数据库 │ │ ↓ │ │ ② 【执行Python代码】 │ │ 批量创建节点公司/人物/产品 │ │ ↓ │ │ ③ 【执行Python代码】 │ │ 批量创建关系持股/任职/供应/竞争 │ │ ↓ │ │ ④ 【执行Python代码】 │ │ 执行Cypher查询做关联分析 │ │ ↓ │ │ ⑤ 【写入Excel】 │ │ 将分析结果输出为报表 │ └──────────────────────────────────────────────┘2.3 连接Neo4j并创建节点步骤①②节点创建fromneo4jimportGraphDatabaseimportjson# Neo4j连接配置 NEO4J_URIbolt://localhost:7687NEO4J_USERneo4jNEO4J_PASSWORDyour_password# 创建驱动 driverGraphDatabase.driver(NEO4J_URI,auth(NEO4J_USER,NEO4J_PASSWORD))defcreate_company_node(tx,name,industry,registered_capital):创建公司节点query MERGE (c:Company {name: $name}) SET c.industry $industry, c.registered_capital $registered_capital, c.updated_at datetime() RETURN c resulttx.run(query,namename,industryindustry,registered_capitalregistered_capital)returnresult.single()defcreate_person_node(tx,name,title,phone):创建人物节点query MERGE (p:Person {name: $name}) SET p.title $title, p.phone $phone, p.updated_at datetime() RETURN p resulttx.run(query,namename,titletitle,phonephone)returnresult.single()# 批量创建节点 # 从Excel读取的数据这里用示例数据实际从影刀Excel节点获取companies[{name:山海科技有限公司,industry:信息技术,registered_capital:1000万},{name:星辰贸易有限公司,industry:批发零售,registered_capital:500万},{name:云帆投资控股,industry:金融投资,registered_capital:5000万},]persons[{name:张三,title:总经理,phone:13800001111},{name:李四,title:法人代表,phone:13800002222},{name:王五,title:监事,phone:13800003333},]withdriver.session()assession:forcompanyincompanies:session.execute_write(create_company_node,company[name],company[industry],company[registered_capital],)print(f创建公司节点{company[name]})forpersoninpersons:session.execute_write(create_person_node,person[name],person[title],person[phone],)print(f创建人物节点{person[name]})print(节点创建完成)2.4 创建关系步骤③建立节点之间的关联defcreate_shareholding_relation(tx,shareholder,company,ratio):创建持股关系query MATCH (p:Person {name: $shareholder}) MATCH (c:Company {name: $company}) MERGE (p)-[r:持股]-(c) SET r.ratio $ratio, r.updated_at datetime() RETURN r resulttx.run(query,shareholdershareholder,companycompany,ratioratio)returnresult.single()defcreate_supply_relation(tx,supplier,customer,product):创建供应关系query MATCH (s:Company {name: $supplier}) MATCH (c:Company {name: $customer}) MERGE (s)-[r:供应]-(c) SET r.product $product, r.updated_at datetime() RETURN r resulttx.run(query,suppliersupplier,customercustomer,productproduct)returnresult.single()defcreate_employment_relation(tx,person,company,position):创建任职关系query MATCH (p:Person {name: $person}) MATCH (c:Company {name: $company}) MERGE (p)-[r:任职于]-(c) SET r.position $position, r.updated_at datetime() RETURN r resulttx.run(query,personperson,companycompany,positionposition)returnresult.single()# 批量创建关系 relations[# (类型, 源, 目标, 附加属性)(shareholding,张三,山海科技有限公司,60%),(shareholding,李四,山海科技有限公司,40%),(shareholding,云帆投资控股,星辰贸易有限公司,80%),(employment,张三,山海科技有限公司,总经理),(employment,李四,山海科技有限公司,法人),(employment,王五,星辰贸易有限公司,监事),(supply,山海科技有限公司,星辰贸易有限公司,ERP系统),]withdriver.session()assession:forrel_type,src,dst,attrinrelations:ifrel_typeshareholding:session.execute_write(create_shareholding_relation,src,dst,attr)print(f创建持股关系{src}→{dst}({attr}))elifrel_typeemployment:session.execute_write(create_employment_relation,src,dst,attr)print(f创建任职关系{src}→{dst}({attr}))elifrel_typesupply:session.execute_write(create_supply_relation,src,dst,attr)print(f创建供应关系{src}→{dst}({attr}))print(关系创建完成)2.5 Cypher查询做关联分析步骤④核心分析查询defquery_shareholding_chain(tx,company_name,depth3):查询股权穿透链向上追溯N层股东queryf MATCH path (shareholder)-[r:持股*1..{depth}]-(c:Company {{name: $company_name}}) RETURN shareholder.name AS 股东, [rel IN relationships(path) | rel.ratio] AS 持股比例, length(path) AS 穿透层数 ORDER BY 穿透层数 resulttx.run(query,company_namecompany_name)return[record.data()forrecordinresult]defquery_related_persons(tx,company_name):查询与公司关联的所有人员任职持股query MATCH (p:Person)-[r]-(c:Company {name: $company_name}) WHERE type(r) IN [任职于, 持股] RETURN p.name AS 姓名, type(r) AS 关系类型, COALESCE(r.position, r.ratio) AS 详情 resulttx.run(query,company_namecompany_name)return[record.data()forrecordinresult]defquery_supply_chain(tx,company_name):查询供应链上下游query MATCH (upstream)-[r1:供应]-(c:Company {name: $company_name}) OPTIONAL MATCH (c)-[r2:供应]-(downstream) RETURN c.name AS 当前公司, upstream.name AS 上游供应商, r1.product AS 供应产品, downstream.name AS 下游客户, r2.product AS 输出产品 resulttx.run(query,company_namecompany_name)return[record.data()forrecordinresult]# 执行查询 analysis_results[]withdriver.session()assession:# 1. 股权穿透chainsession.execute_read(query_shareholding_chain,星辰贸易有限公司)analysis_results.append({分析类型:股权穿透,数据:chain})# 2. 关联人员persons_relatedsession.execute_read(query_related_persons,山海科技有限公司)analysis_results.append({分析类型:关联人员,数据:persons_related})# 3. 供应链supplysession.execute_read(query_supply_chain,山海科技有限公司)analysis_results.append({分析类型:供应链,数据:supply})# 输出结果供影刀Excel写入outputjson.dumps(analysis_results,ensure_asciiFalse)driver.close()2.6 写入Excel报告影刀中【遍历列表】→ 遍历analysis_results 【写入Excel】→ 每个分析类型写一个Sheet页 Sheet1「股权穿透」→ 写入chain数据 Sheet2「关联人员」→ 写入persons_related数据  Sheet3「供应链」→ 写入supply数据三、有什么坑坑1MERGE vs CREATE的区别新手最容易搞混的两个操作CREATE无条件创建即使节点/关系已经存在也会创建一份新的产生重复MERGE如果不存在则创建如果存在则匹配已有节点数据更新的场景一定用MERGE-- ✅ 正确合并添加不会重复 MERGE (c:Company {name: 山海科技}) -- ❌ 错误每执行一次就多一条重复记录 CREATE (c:Company {name: 山海科技})坑2Neo4j连接不关闭导致资源泄漏影刀流程如果报错中途退出driver对象不会被关闭Neo4j服务端会积累大量空闲连接。TEMU店群矩阵自动化运营核价报活动确保关闭# 方案1with语句推荐自动关闭withGraphDatabase.driver(...)asdriver:withdriver.session()assession:# ... 操作# 方案2手动try-finallydriverGraphDatabase.driver(...)try:# ... 操作finally:driver.close()坑3Cypher注入风险如果你在Cypher语句里用字符串拼接拼接用户输入比如公司名有注入风险# ❌ 危险字符串拼接queryfMATCH (c:Company {{name: {company_name}}}) RETURN c# ✅ 安全参数化查询queryMATCH (c:Company {name: $company_name}) RETURN ctx.run(query,company_namecompany_name)坑4深度穿透查询性能差查询3层以上的股权穿透数据量一大就慢。建议限制最大深度不超过5层对大图建立索引CREATE INDEX FOR (c:Company) ON (c.name)考虑在Neo4j中预先计算路径而不是实时查询坑5影刀Python节点中安装neo4j驱动失败neo4j驱动依赖一些系统库Windows下偶尔安装失败。如果遇到# 尝试安装旧版本兼容性更好pipinstallneo4j4.4.0# 或者用py2neo另一个驱动库更轻量pipinstallpy2neopy2neo的API不同但思路一样文档很全。总结影刀 Neo4j 的核心价值在于「自动化采集 图分析」的组合。影刀负责从各种地方网页、Excel、API把关系数据抓回来Neo4j负责存储和分析关系。MERGE别写成CREATE查询一定要参数化连接用完记得关。