From c0ab45ab81685384a1c8679a93153423a3a378dd Mon Sep 17 00:00:00 2001
From: yumoqing <yumoqing@gmail.com>
Date: Sat, 4 Jul 2026 20:11:17 +0800
Subject: [PATCH] feat: initial commit - graph-service

---
 .gitignore              |  10 ++
 ah.py                   |   6 ++
 build.sh                |  55 +++++++++++
 conf/config.json        |  33 +++++++
 data/default.json       |   7 ++
 data/test.json          |  29 ++++++
 init.py                 | 184 ++++++++++++++++++++++++++++++++++++
 workers/__init__.py     |   0
 workers/graph_engine.py | 203 ++++++++++++++++++++++++++++++++++++++++
 9 files changed, 527 insertions(+)
 create mode 100644 .gitignore
 create mode 100644 ah.py
 create mode 100755 build.sh
 create mode 100644 conf/config.json
 create mode 100644 data/default.json
 create mode 100644 data/test.json
 create mode 100644 init.py
 create mode 100644 workers/__init__.py
 create mode 100644 workers/graph_engine.py

diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..5cbdaeb
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,10 @@
+__pycache__/
+*.pyc
+*.pyo
+logs/
+*.log
+nohup.out
+nohup_gpu*.out
+py3/
+*.egg-info/
+*.pid
diff --git a/ah.py b/ah.py
new file mode 100644
index 0000000..efe61a2
--- /dev/null
+++ b/ah.py
@@ -0,0 +1,6 @@
+# -*- coding:utf-8 -*-
+from ahserver.webapp import webapp
+from init import load_graph_service
+
+if __name__ == '__main__':
+    webapp(load_graph_service)
diff --git a/build.sh b/build.sh
new file mode 100755
index 0000000..92b24f4
--- /dev/null
+++ b/build.sh
@@ -0,0 +1,55 @@
+#!/usr/bin/env bash
+# Graph Service (NetworkX)
+set -e
+cd "$(dirname "$0")"
+
+SERVICE_NAME="graph-service"
+PORT=9092
+PY=/data/ymq/wan22-service/py3/bin/python
+action="${1:-status}"
+
+case "$action" in
+    deploy|update)
+        echo "=== $SERVICE_NAME Deploy (CPU, port $PORT) ==="
+        if [ -f ah.pid ] && kill -0 $(cat ah.pid) 2>/dev/null; then
+            kill $(cat ah.pid) 2>/dev/null || true; sleep 2
+        fi
+        if [ -d .git ] && [ -f .git/HEAD ]; then
+            git pull origin master 2>/dev/null || true
+        fi
+        mkdir -p logs files wwwroot data
+        export PYTHONPATH="$(pwd)"
+        nohup $PY ah.py > nohup.out 2>&1 &
+        echo $! > ah.pid
+        echo "Started PID $(cat ah.pid) on port $PORT"
+        sleep 3
+        if curl -s http://localhost:$PORT/api/status > /dev/null 2>&1; then
+            echo "Service healthy"
+        else
+            echo "WARNING: not responding, check nohup.out"
+            tail -20 nohup.out
+        fi
+        ;;
+    stop)
+        if [ -f ah.pid ]; then
+            kill $(cat ah.pid) 2>/dev/null || true; rm -f ah.pid; echo "Stopped"
+        else echo "Not running"; fi
+        ;;
+    start)
+        mkdir -p logs files wwwroot data
+        export PYTHONPATH="$(pwd)"
+        nohup $PY ah.py > nohup.out 2>&1 &
+        echo $! > ah.pid; echo "Started PID $(cat ah.pid)"
+        ;;
+    status)
+        echo "=== $SERVICE_NAME Status ==="
+        if [ -f ah.pid ] && kill -0 $(cat ah.pid) 2>/dev/null; then
+            echo "Process: running (PID $(cat ah.pid))"
+        else echo "Process: not running"; fi
+        echo "Port: $PORT"
+        if curl -s --max-time 3 http://localhost:$PORT/api/status > /dev/null 2>&1; then
+            echo "HTTP: OK"
+        else echo "HTTP: not responding"; fi
+        ;;
+    *) echo "Usage: $0 {deploy|update|stop|start|status}"; exit 1 ;;
+esac
diff --git a/conf/config.json b/conf/config.json
new file mode 100644
index 0000000..264848f
--- /dev/null
+++ b/conf/config.json
@@ -0,0 +1,33 @@
+{
+    "password_key": "GraphService2026Key",
+    "filesroot": "$[workdir]$/files",
+    "logger": {
+        "name": "graph-service",
+        "levelname": "info",
+        "logfile": "$[workdir]$/logs/graph-service.log"
+    },
+    "website": {
+        "paths": [["$[workdir]$/wwwroot", ""]],
+        "client_max_size": 52428800,
+        "host": "0.0.0.0",
+        "port": 9092,
+        "coding": "utf-8",
+        "indexes": ["index.html"],
+        "startswiths": [
+            {"leading": "/api/status", "registerfunction": "status"},
+            {"leading": "/api/debug", "registerfunction": "debug"},
+            {"leading": "/api/graph/add_node", "registerfunction": "add_node"},
+            {"leading": "/api/graph/add_edge", "registerfunction": "add_edge"},
+            {"leading": "/api/graph/neighbors", "registerfunction": "neighbors"},
+            {"leading": "/api/graph/path", "registerfunction": "path"},
+            {"leading": "/api/graph/query", "registerfunction": "query"},
+            {"leading": "/api/graph/stats", "registerfunction": "stats"},
+            {"leading": "/api/graph/save", "registerfunction": "save"},
+            {"leading": "/api/graph/load", "registerfunction": "load"}
+        ],
+        "processors": [
+            [".tmpl", "tmpl"], [".app", "app"], [".ui", "bui"],
+            [".dspy", "dspy"], [".md", "md"]
+        ]
+    }
+}
diff --git a/data/default.json b/data/default.json
new file mode 100644
index 0000000..cd7f495
--- /dev/null
+++ b/data/default.json
@@ -0,0 +1,7 @@
+{
+  "directed": true,
+  "multigraph": false,
+  "graph": {},
+  "nodes": [],
+  "links": []
+}
\ No newline at end of file
diff --git a/data/test.json b/data/test.json
new file mode 100644
index 0000000..e94275e
--- /dev/null
+++ b/data/test.json
@@ -0,0 +1,29 @@
+{
+  "directed": true,
+  "multigraph": false,
+  "graph": {},
+  "nodes": [
+    {
+      "name": "张三",
+      "age": 30,
+      "type": "person",
+      "id": "person_1"
+    },
+    {
+      "id": "person_2"
+    },
+    {
+      "id": "company_1"
+    }
+  ],
+  "links": [
+    {
+      "source": "person_1",
+      "target": "person_2"
+    },
+    {
+      "source": "person_2",
+      "target": "company_1"
+    }
+  ]
+}
\ No newline at end of file
diff --git a/init.py b/init.py
new file mode 100644
index 0000000..1e9a971
--- /dev/null
+++ b/init.py
@@ -0,0 +1,184 @@
+# -*- coding:utf-8 -*-
+from traceback import format_exc
+from ahserver.serverenv import ServerEnv
+from appPublic.registerfunction import RegisterFunction
+from appPublic.log import exception
+import json
+
+
+async def status_handler(request, params_kw, *args, **kwargs):
+    import sys, os
+    sys.path.insert(0, os.getcwd())
+    from workers.graph_engine import health_check
+    health = health_check()
+    return json.dumps({
+        "service": "graph-service",
+        "backend": health["backend"],
+        "graphs_loaded": health["graphs_loaded"],
+        "data_dir": health["data_dir"],
+        "endpoints": [
+            "/api/status", "/api/graph/add_node", "/api/graph/add_edge",
+            "/api/graph/neighbors", "/api/graph/path", "/api/graph/query",
+            "/api/graph/stats", "/api/graph/save", "/api/graph/load"
+        ]
+    }, indent=2, ensure_ascii=False)
+
+
+async def debug_handler(request, params_kw, *args, **kwargs):
+    """Debug endpoint to see params_kw contents"""
+    return json.dumps({
+        "params_kw": params_kw,
+        "params_kw_keys": list(params_kw.keys()),
+        "request_method": request.method,
+        "request_content_type": request.content_type
+    }, indent=2, ensure_ascii=False)
+
+
+async def add_node_handler(request, params_kw, *args, **kwargs):
+    import sys, os
+    sys.path.insert(0, os.getcwd())
+    from workers.graph_engine import add_node
+    try:
+        graph = params_kw.get("graph", "default")
+        node_id = params_kw.get("node_id")
+        attrs = params_kw.get("attrs", {})
+        
+        if not node_id:
+            return json.dumps({"error": "node_id required", "debug": {"params_kw": params_kw}})
+        
+        result = add_node(graph, node_id, **attrs)
+        return json.dumps({"status": "SUCCEEDED", **result}, ensure_ascii=False)
+    except Exception as e:
+        exception(f"{e}, {format_exc()}")
+        return json.dumps({"error": str(e)})
+
+
+async def add_edge_handler(request, params_kw, *args, **kwargs):
+    import sys, os
+    sys.path.insert(0, os.getcwd())
+    from workers.graph_engine import add_edge
+    try:
+        graph = params_kw.get("graph", "default")
+        source = params_kw.get("source")
+        target = params_kw.get("target")
+        attrs = params_kw.get("attrs", {})
+        
+        if not source or not target:
+            return json.dumps({"error": "source and target required"})
+        
+        result = add_edge(graph, source, target, **attrs)
+        return json.dumps({"status": "SUCCEEDED", **result}, ensure_ascii=False)
+    except Exception as e:
+        exception(f"{e}, {format_exc()}")
+        return json.dumps({"error": str(e)})
+
+
+async def neighbors_handler(request, params_kw, *args, **kwargs):
+    import sys, os
+    sys.path.insert(0, os.getcwd())
+    from workers.graph_engine import get_neighbors
+    try:
+        graph = params_kw.get("graph", "default")
+        node_id = params_kw.get("node_id")
+        direction = params_kw.get("direction", "both")
+        depth = int(params_kw.get("depth", 1))
+        
+        if not node_id:
+            return json.dumps({"error": "node_id required"})
+        
+        result = get_neighbors(graph, node_id, direction, depth)
+        return json.dumps({"status": "SUCCEEDED", **result}, ensure_ascii=False)
+    except Exception as e:
+        exception(f"{e}, {format_exc()}")
+        return json.dumps({"error": str(e)})
+
+
+async def path_handler(request, params_kw, *args, **kwargs):
+    import sys, os
+    sys.path.insert(0, os.getcwd())
+    from workers.graph_engine import find_path
+    try:
+        graph = params_kw.get("graph", "default")
+        source = params_kw.get("source")
+        target = params_kw.get("target")
+        max_depth = int(params_kw.get("max_depth", 10))
+        
+        if not source or not target:
+            return json.dumps({"error": "source and target required"})
+        
+        result = find_path(graph, source, target, max_depth)
+        return json.dumps({"status": "SUCCEEDED", **result}, ensure_ascii=False)
+    except Exception as e:
+        exception(f"{e}, {format_exc()}")
+        return json.dumps({"error": str(e)})
+
+
+async def query_handler(request, params_kw, *args, **kwargs):
+    import sys, os
+    sys.path.insert(0, os.getcwd())
+    from workers.graph_engine import query_nodes
+    try:
+        graph = params_kw.get("graph", "default")
+        filters = params_kw.get("filters")
+        limit = int(params_kw.get("limit", 100))
+        
+        result = query_nodes(graph, filters, limit)
+        return json.dumps({"status": "SUCCEEDED", **result}, ensure_ascii=False)
+    except Exception as e:
+        exception(f"{e}, {format_exc()}")
+        return json.dumps({"error": str(e)})
+
+
+async def stats_handler(request, params_kw, *args, **kwargs):
+    import sys, os
+    sys.path.insert(0, os.getcwd())
+    from workers.graph_engine import get_stats
+    try:
+        graph = params_kw.get("graph", "default")
+        result = get_stats(graph)
+        return json.dumps({"status": "SUCCEEDED", **result}, ensure_ascii=False)
+    except Exception as e:
+        exception(f"{e}, {format_exc()}")
+        return json.dumps({"error": str(e)})
+
+
+async def save_handler(request, params_kw, *args, **kwargs):
+    import sys, os
+    sys.path.insert(0, os.getcwd())
+    from workers.graph_engine import save_graph
+    try:
+        graph = params_kw.get("graph", "default")
+        result = save_graph(graph)
+        return json.dumps({"status": "SUCCEEDED", **result}, ensure_ascii=False)
+    except Exception as e:
+        exception(f"{e}, {format_exc()}")
+        return json.dumps({"error": str(e)})
+
+
+async def load_handler(request, params_kw, *args, **kwargs):
+    import sys, os
+    sys.path.insert(0, os.getcwd())
+    from workers.graph_engine import load_graph
+    try:
+        graph = params_kw.get("graph", "default")
+        result = load_graph(graph)
+        return json.dumps({"status": "SUCCEEDED", **result}, ensure_ascii=False)
+    except Exception as e:
+        exception(f"{e}, {format_exc()}")
+        return json.dumps({"error": str(e)})
+
+
+def load_graph_service():
+    """Register API handlers"""
+    env = ServerEnv()
+    rf = RegisterFunction()
+    rf.register("status", status_handler)
+    rf.register("debug", debug_handler)
+    rf.register("add_node", add_node_handler)
+    rf.register("add_edge", add_edge_handler)
+    rf.register("neighbors", neighbors_handler)
+    rf.register("path", path_handler)
+    rf.register("query", query_handler)
+    rf.register("stats", stats_handler)
+    rf.register("save", save_handler)
+    rf.register("load", load_handler)
diff --git a/workers/__init__.py b/workers/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/workers/graph_engine.py b/workers/graph_engine.py
new file mode 100644
index 0000000..d02c664
--- /dev/null
+++ b/workers/graph_engine.py
@@ -0,0 +1,203 @@
+# -*- coding:utf-8 -*-
+"""NetworkX-based graph engine with persistence."""
+import os
+import json
+import networkx as nx
+from typing import Dict, List, Any, Optional
+
+DATA_DIR = "/data/ymq/graph-service/data"
+_graphs: Dict[str, nx.DiGraph] = {}
+
+
+def get_graph(graph_name: str = "default") -> nx.DiGraph:
+    """Get or create a graph by name."""
+    if graph_name not in _graphs:
+        # Try to load from disk
+        filepath = os.path.join(DATA_DIR, f"{graph_name}.json")
+        if os.path.exists(filepath):
+            with open(filepath, 'r') as f:
+                data = json.load(f)
+            _graphs[graph_name] = nx.node_link_graph(data, directed=True)
+        else:
+            _graphs[graph_name] = nx.DiGraph()
+    return _graphs[graph_name]
+
+
+def add_node(graph_name: str, node_id: str, **attrs) -> Dict:
+    """Add a node to the graph."""
+    g = get_graph(graph_name)
+    g.add_node(node_id, **attrs)
+    return {"status": "ok", "node_id": node_id, "graph": graph_name}
+
+
+def add_edge(graph_name: str, source: str, target: str, **attrs) -> Dict:
+    """Add an edge to the graph."""
+    g = get_graph(graph_name)
+    g.add_edge(source, target, **attrs)
+    return {"status": "ok", "source": source, "target": target, "graph": graph_name}
+
+
+def get_neighbors(graph_name: str, node_id: str, direction: str = "both", depth: int = 1) -> Dict:
+    """Get neighbors of a node."""
+    g = get_graph(graph_name)
+    
+    if node_id not in g:
+        return {"error": f"Node {node_id} not found"}
+    
+    # BFS to get neighbors up to depth
+    visited = set()
+    queue = [(node_id, 0)]
+    neighbors = []
+    
+    while queue:
+        current, d = queue.pop(0)
+        if d > depth:
+            break
+        
+        if d > 0:  # Don't include the starting node
+            neighbors.append({
+                "node_id": current,
+                "depth": d,
+                "attrs": dict(g.nodes[current])
+            })
+        
+        if current not in visited:
+            visited.add(current)
+            
+            if d < depth:
+                # Add successors
+                if direction in ["both", "out"]:
+                    for neighbor in g.successors(current):
+                        if neighbor not in visited:
+                            queue.append((neighbor, d + 1))
+                
+                # Add predecessors
+                if direction in ["both", "in"]:
+                    for neighbor in g.predecessors(current):
+                        if neighbor not in visited:
+                            queue.append((neighbor, d + 1))
+    
+    return {"node_id": node_id, "neighbors": neighbors, "count": len(neighbors)}
+
+
+def find_path(graph_name: str, source: str, target: str, max_depth: int = 10) -> Dict:
+    """Find shortest path between two nodes."""
+    g = get_graph(graph_name)
+    
+    if source not in g:
+        return {"error": f"Source node {source} not found"}
+    if target not in g:
+        return {"error": f"Target node {target} not found"}
+    
+    try:
+        path = nx.shortest_path(g, source, target)
+        if len(path) > max_depth + 1:
+            return {"error": f"Path too long: {len(path)-1} > {max_depth}"}
+        
+        # Get path details
+        path_details = []
+        for i in range(len(path)):
+            node_id = path[i]
+            node_data = {"node_id": node_id, "attrs": dict(g.nodes[node_id])}
+            
+            if i < len(path) - 1:
+                # Add edge info
+                next_node = path[i + 1]
+                edge_data = dict(g.edges[node_id, next_node])
+                node_data["edge_to_next"] = edge_data
+            
+            path_details.append(node_data)
+        
+        return {
+            "source": source,
+            "target": target,
+            "path": path,
+            "length": len(path) - 1,
+            "details": path_details
+        }
+    except nx.NetworkXNoPath:
+        return {"error": "No path found"}
+
+
+def query_nodes(graph_name: str, filters: Optional[Dict] = None, limit: int = 100) -> Dict:
+    """Query nodes with optional filters."""
+    g = get_graph(graph_name)
+    
+    results = []
+    for node_id, attrs in g.nodes(data=True):
+        # Apply filters
+        match = True
+        if filters:
+            for key, value in filters.items():
+                if attrs.get(key) != value:
+                    match = False
+                    break
+        
+        if match:
+            results.append({"node_id": node_id, "attrs": attrs})
+            if len(results) >= limit:
+                break
+    
+    return {"nodes": results, "count": len(results)}
+
+
+def get_stats(graph_name: str) -> Dict:
+    """Get graph statistics."""
+    g = get_graph(graph_name)
+    
+    stats = {
+        "graph": graph_name,
+        "nodes": g.number_of_nodes(),
+        "edges": g.number_of_edges(),
+        "density": nx.density(g),
+        "is_connected": nx.is_weakly_connected(g) if g.number_of_nodes() > 0 else True,
+    }
+    
+    if g.number_of_nodes() > 0:
+        # Degree statistics
+        in_degrees = [d for _, d in g.in_degree()]
+        out_degrees = [d for _, d in g.out_degree()]
+        
+        stats["avg_in_degree"] = sum(in_degrees) / len(in_degrees)
+        stats["avg_out_degree"] = sum(out_degrees) / len(out_degrees)
+        stats["max_in_degree"] = max(in_degrees)
+        stats["max_out_degree"] = max(out_degrees)
+    
+    return stats
+
+
+def save_graph(graph_name: str) -> Dict:
+    """Save graph to disk."""
+    g = get_graph(graph_name)
+    filepath = os.path.join(DATA_DIR, f"{graph_name}.json")
+    
+    data = nx.node_link_data(g)
+    with open(filepath, 'w') as f:
+        json.dump(data, f, ensure_ascii=False, indent=2)
+    
+    return {"status": "ok", "filepath": filepath, "nodes": g.number_of_nodes(), "edges": g.number_of_edges()}
+
+
+def load_graph(graph_name: str) -> Dict:
+    """Load graph from disk."""
+    filepath = os.path.join(DATA_DIR, f"{graph_name}.json")
+    
+    if not os.path.exists(filepath):
+        return {"error": f"Graph file not found: {filepath}"}
+    
+    with open(filepath, 'r') as f:
+        data = json.load(f)
+    
+    g = nx.node_link_graph(data, directed=True)
+    _graphs[graph_name] = g
+    
+    return {"status": "ok", "nodes": g.number_of_nodes(), "edges": g.number_of_edges()}
+
+
+def health_check():
+    """Check service status."""
+    return {
+        "backend": "NetworkX",
+        "graphs_loaded": len(_graphs),
+        "data_dir": DATA_DIR
+    }