File: rdfdiff.py

package info (click to toggle)
4store 1.1.6%2B20151109-2
  • links: PTS, VCS
  • area: main
  • in suites: buster, stretch
  • size: 82,388 kB
  • sloc: ansic: 65,689; sh: 2,916; perl: 2,245; makefile: 281; python: 213
file content (62 lines) | stat: -rwxr-xr-x 1,618 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
#!/usr/bin/env python
"""
RDF Graph Isomorphism Tester
Author: Sean B. Palmer, inamidst.com
Uses the pyrple algorithm
Requirements: 
   Python2.4+
   http://inamidst.com/proj/rdf/ntriples.py
Usage: ./rdfdiff.py <ntriplesP> <ntriplesQ>
"""

import sys, re, urllib
import ntriples
from ntriples import bNode

ntriples.r_uriref = re.compile(r'<([^\s"<>]+)>')

class Graph(object): 
   def __init__(self, uri): 
      self.triples = set()
      self.parse(uri)

   def parse(self, uri): 
      class Sink(object): 
         def triple(sink, s, p, o): 
            self.triples.add((s, p, o))

      p = ntriples.NTriplesParser(sink=Sink())
      u = urllib.urlopen(uri)
      p.parse(u)
      u.close()

   def __hash__(self): 
      return hash(tuple(sorted(self.hashtriples())))

   def hashtriples(self): 
      for triple in self.triples: 
         g = ((isinstance(t, bNode) and self.vhash(t)) or t for t in triple)
         yield hash(tuple(g))

   def vhash(self, term, done=False): 
      return tuple(sorted(self.vhashtriples(term, done)))

   def vhashtriples(self, term, done): 
      for t in self.triples: 
         if term in t: yield tuple(self.vhashtriple(t, term, done))

   def vhashtriple(self, triple, term, done): 
      for p in xrange(3): 
         if not isinstance(triple[p], bNode): yield triple[p]
         elif done or (triple[p] == term): yield p
         else: yield self.vhash(triple[p], done=True)

def compare(p, q): 
   return hash(Graph(p)) == hash(Graph(q))

def main(): 
   result = compare(sys.argv[1], sys.argv[2])
   print ('no', 'yes')[result]

if __name__=="__main__": 
   main()