Python 下 Flask 结合 Neo4j 构建管理系统（三）

发表于 2017-04-13 更新于 2022-04-22 分类于 Archived 阅读次数： Disqus：本文字数： 593 阅读时长 ≈ 1 分钟

需求

核心需求如下： * 管理系统登录 * 从 Neo4j 中获取数据信息图形化显示出来 * 节点的关系查询及溯源查询 * 对 Neo4j 进行节点的增删改

管理系统登录

首先需要在工程目录下创建 login.html 和 models.py，前者为登录界面用于测试与 Neo4j 的连接，后者是一个用户类用于处理用户的各项操作。

本次连接采用了 Py2neo v3 库，安装可以通过 pip 安装：pip install py2neo
安装成功之后在 models.py 文件输入如下代码：

from passlib.handlers import bcrypt
from py2neo import authenticate, Graph, Node

authenticate("localhost:7474", "neo4j", "yy0758jyl")
graph = Graph()


class User:
    def __init__(self, username):
        self.username = username
        user_data = self.find()
        #获取登录用户标签
        if user_data:
            self.label = user_data['label']
        else:
            self.label = "User"

    def find(self):
        user = {}
        data_user = graph.find_one("User", "username", self.username)
        if not data_user:
            data_admin = graph.find_one("Admin", "username", self.username)
            if not data_admin:
                return False
            else:
                print(user)
                user['username'] = data_admin['username']
                user['password'] = data_admin['password']
                user['label'] = 'Admin'
        else:
            user['username'] = data_user['username']
            user['password'] = data_user['password']
            user['label'] = 'User'
        return user

    def register(self, password):
        if not self.find():
            user = Node("User", username=self.username, password=bcrypt.encrypt(password))
            graph.create(user)
            cypher = 'MATCH (admin:Admin) MATCH(user:User {username:\'' + self.username + '\'}) CREATE (admin)-[manage:Manage]->(user)'
            graph.run(
                'MATCH (admin:Admin) MATCH(user:User {username:\'' + self.username + '\'}) CREATE (admin)-[manage:Manage]->(user)')
            return True
        else:
            return False

    def verify_pass(self, password):
        user = self.find()
        if not user:
            return False
        else:
            return bcrypt.verify(password, user["password"])

authenticate() 用于连接 Neo4j 数据库，三个参数依次是地址，用户名，密码。
find(self) 用户查找用户。
register(self, password) 注册用户。
verify_pass(self, password) 登录验证。

app.py 代码：

app = Flask(__name__)
app.secret_key = 'sugarsugar'

@app.route('/', methods=["GET", "POST"])
def index():
    if request.method == "POST":
        if "registeruser" in request.form:
            username = request.form['registeruser']
            password = request.form['password']
            if not username or not password:
                flash('请输入用户名和密码')
            else:
                user = User(username)
                print(user)
                if not user.register(password):
                    flash('一个用户已经存在')
                else:
                    flash('注册成功')
        else:
            username = request.form['username']
            password = request.form['password']

            user = User(username)
            if not username or not password:
                flash('请输入用户名和密码')
            elif not user.verify_pass(password):
                flash('无效登录')
            else:
                session["username"] = user.username
                session['label'] = user.label
                return redirect(url_for("main_view"))
    return render_template('login.html')

app.secret_key 由于使用了 Session 所以必须要设置该项值。
flash() 用于向网页发送信息。

数据查询

在测试完登录功能之后，首先在 templates 文件夹下创建 view.html 文件，作为登录后跳转的页面：主页面执行查询后相应的前端代码：

$(document).ready(function () {
    function search() {
        var query = $("#search").find("input[name=condition]").val();
        var flag = $("#search").find("input[name=trace]").prop('checked');
        if(flag == true && query == "ebay") {
            alert("不可为根节点");
            return false;
        }
        $.get("/search" , {q: query, f : flag},
            function (data) {
                if(data.nodes.length > 0){
                    update(data.nodes, data.links);
                    update_static(data.nodes);
                }
                else
                    alert("查询结果为空");
            }, "json");
        return false;
    }

    $("#search").submit(search);
    search();
});

后台处理代码：

app.py 代码：

@app.route('/search')
def get_search():
  try:
      q = request.args.get('q', 0)
      f = request.args.get('f', 1)
  except KeyError:
      return []
  else:
      user = User(session.get("username"))
      results = user.search_graph(q, f)
      return results

models.py 代码：

def search_graph(self, condition, flag=False):
    nodes = []
    rels = []
    i = 0
    #关系查询
    query = "MATCH (m)<-[:BELONG]-(a) WHERE m.name CONTAINS \'" + condition + "\' or a.name CONTAINS \'" + condition + \
            "\' RETURN properties(m) as start_node, labels(m) as labels, " \
            "collect(properties(a)) as cast,collect(labels(a)) as cast_labels"
    #溯源查询
    query_trace = "Match (n) WHERE n.name contains \'" + condition + \
                  "\' MATCH path = shortestPath( (m:Website)-[:BELONG*..5]-(n) )" \
                  " where labels(n)='Sort' or labels(n)='GOODS'" \
                  " RETURN nodes(path) AS nodes, EXTRACT(node IN nodes(path) | ID(node)) AS ids " \
                  ",EXTRACT(node IN nodes(path) | labels(node)) AS labels"
    results = graph.run(query)
    if flag == "true" and condition:
        nodes_act = []
        rels_act = []
        for node_trace, id_trace, label_trace in graph.run(query_trace):
            j = 0
            for node_ori in node_trace:
                node_act = {}
                node_keys = list(node_ori.keys())
                for key in node_keys:
                    node_act[key] = node_ori.get(key)
                node_act['label'] = label_trace[j][0]
                try:
                    target_act = nodes_act.index(node_act)
                    j += 1
                except ValueError:
                    target_act = len(nodes_act)
                    nodes_act.append(node_act)
                    j += 1
                if j < len(node_trace):
                    try:
                        source_act = nodes_act.index(self.get_node(node_trace[j], label_trace[j]))
                    except ValueError:
                        source_act = len(nodes_act)
                    rels_act.append({"source": source_act, "target": target_act})
        return jsonify({"nodes": nodes_act, "links": rels_act})

    for start_nodes, start_labels, end_nodes, end_labels in results:
        keys = list(start_nodes.keys())
        node = {}
        for key in keys:
            node[key] = start_nodes.get(key)
        node['label'] = start_labels[0]
        try:
            target = nodes.index(node)
        except ValueError:
            nodes.append(node)
            target = i
            i += 1
        for end_node, label in zip(end_nodes, end_labels):
            end_keys = list(end_node.keys())
            end = {}
            for key in end_keys:
                end[key] = end_node.get(key)
            end['label'] = label[0]
            try:
                source = nodes.index(end)
            except ValueError:
                nodes.append(end)
                source = i
                i += 1
            rels.append({"source": source, "target": target})
    return jsonify({"nodes": nodes, "links": rels})

数据的可视化采用 d3.js，具体过程参考d3.js 绘制力导向图，返回的数据格式限定为 Json 格式，nodes 保存节点信息，links 保存节点关系：

1
2

{"nodes": [{'label': 'Website', 'name': 'ebay'}, {'label': 'Sort', 'name': 'Health & Beauty'}, {'label': 'Sort', 'name': 'Musical Instruments & Gear'}],
 "links":[{'target': 0, 'source': 1}, {'target': 0, 'source': 2}]}

查询效果：

参考资料

Py2neo v3 Handbook