Skip to content

Instantly share code, notes, and snippets.

@edsu edsu/chronam-rdf.py
Created May 22, 2009

Embed
What would you like to do?
some code to turn some django models into rdf for the chroniclingamerica.loc.gov application
from rdflib import ConjunctiveGraph, Namespace, Literal, URIRef, RDF, RDFS
from chronam.utils.rfc3339 import rfc3339
DC = Namespace('http://purl.org/dc/elements/1.1/')
ORE = Namespace('http://www.openarchives.org/ore/terms/')
OWL = Namespace('http://www.w3.org/2002/07/owl#')
RDA = Namespace('http://rdvocab.info/elements/')
XSD = Namespace('http://www.w3.org/2001/XMLSchema#')
BIBO = Namespace('http://purl.org/ontology/bibo/')
FOAF = Namespace('http://xmlns.com/foaf/0.1/')
NDNP = Namespace('http://chroniclingamerica.loc.gov/terms#')
DCTERMS = Namespace('http://purl.org/dc/terms/')
def title_to_graph(t, g=None, include_issues=True):
if not g:
g = make_graph()
uri = abstract_uri(t, 'title')
g.add((uri, RDF.type, BIBO['Newspaper']))
g.add((uri, DCTERMS['title'], Literal(t.name)))
g.add((uri, RDA['placeOfPublication'], Literal(t.place_of_publication)))
for subject in t.subjects.all():
g.add((uri, DC['subject'], Literal(subject.heading)))
for language in t.languages.all():
if language.lingvoj:
g.add((uri, DCTERMS['language'], URIRef(language.lingvoj)))
if t.publisher and t.publisher != 's.n.':
g.add((uri, DC['publisher'], Literal(t.publisher)))
if t.essay:
g.add((uri, DCTERMS['description'], URIRef(t.essay.url)))
if include_issues:
for issue in t.issues.all():
g.add((uri, ORE['aggregates'], abstract_uri(issue, 'issue')))
for url in t.urls.all():
g.add((uri, RDFS.seeAlso, URIRef(url.value)))
for place in t.places.all():
if place.dbpedia:
g.add((uri, DCTERMS['coverage'], URIRef(place.dbpedia)))
if place.geonames:
g.add((uri, DCTERMS['coverage'], URIRef(place.geonames)))
if t.start_year and t.end_year:
start, end = t.start_year, t.end_year
if end == 'current':
end = 'unknown'
if start == '????':
start = 'unknown'
g.add((uri, DCTERMS['date'], Literal('%s/%s' % (start, end),
datatype=URIRef('http://www.loc.gov/standards/datetime#edt'))))
g.add((uri, ORE['isDescribedBy'], URIRef(rdf_uri(t))))
g.add((uri, RDFS.seeAlso, URIRef(t.marc.url)))
g.add((uri, RDFS.seeAlso, URIRef('http://lccn.loc.gov/%s' % t.lccn)))
g.add((uri, OWL['sameAs'], URIRef('info:lccn/%s' % t.lccn)))
if t.oclc:
g.add((uri, RDFS.seeAlso,
URIRef('http://www.worldcat.org/oclc/%s' % t.oclc)))
g.add((uri, OWL['sameAs'], URIRef('info:oclcnum/%s' % t.oclc)))
if t.issn:
g.add((uri, OWL['sameAs'], URIRef('urn:issn:%s' % t.issn)))
return g
def issue_to_graph(i, g=None):
if not g:
g = make_graph()
uri = abstract_uri(i, 'issue')
g.add((uri, RDF.type, BIBO['Issue']))
g.add((uri, DCTERMS['title'], Literal('%s - %s' % (i.title.name,
i.date_issued))))
g.add((uri, DCTERMS['issued'], Literal(i.date_issued, datatype=XSD.date)))
g.add((uri, ORE['isAggregatedBy'], abstract_uri(i.title, 'title')))
g.add((uri, ORE['isAggregatedBy'], abstract_uri(i.batch, 'batch')))
for page in i.pages.all():
g.add((uri, ORE['aggregates'], abstract_uri(page, 'page')))
g.add((uri, ORE['isDescribedBy'], URIRef(rdf_uri(i))))
return g
def page_to_graph(p, g=None):
if not g:
g = make_graph()
uri = abstract_uri(p, 'page')
g.add((uri, RDF.type, NDNP['Page']))
g.add((uri, NDNP['sequence'], Literal(p.sequence)))
g.add((uri, ORE['isAggregatedBy'], abstract_uri(p.issue, 'issue')))
jp2_uri = URIRef(p.jp2_url)
g.add((uri, ORE['aggregates'], jp2_uri))
g.add((jp2_uri, DC['format'], Literal('image/jp2')))
ocr_uri = URIRef(p.ocr_url)
g.add((uri, ORE.aggregates, ocr_uri))
g.add((ocr_uri, DC['format'], Literal('text/xml')))
pdf_uri = URIRef(p.pdf_url)
g.add((uri, ORE.aggregates, pdf_uri))
g.add((pdf_uri, DC['format'], Literal('application/pdf')))
txt_uri = URIRef(p.txt_url)
g.add((uri, ORE.aggregates, txt_uri))
g.add((txt_uri, DC['format'], Literal('text/plain')))
thumb_uri = URIRef(p.thumb_url)
g.add((uri, ORE.aggregates, thumb_uri))
g.add((uri, FOAF.depiction, thumb_uri))
g.add((thumb_uri, DC['format'], Literal('image/jpeg')))
g.add((uri, DCTERMS['issued'], Literal(p.issue.date_issued,
datatype=XSD.date)))
g.add((uri, DCTERMS['title'], Literal('%s - %s - %s' %
(p.issue.title.name, p.issue.date_issued, p.sequence))))
if p.number:
g.add((uri, NDNP['number'], Literal(p.number)))
if p.section_label:
g.add((uri, NDNP['section'], Literal(p.section_label)))
g.add((uri, ORE['isDescribedBy'], URIRef(rdf_uri(p))))
return g
def titles_to_graph(titles):
g = make_graph()
for title in titles:
g = title_to_graph(title, g, include_issues=False)
return g
def batch_to_graph(b):
g = make_graph()
uri = abstract_uri(b, 'batch')
g.add((uri, RDF.type, NDNP['Batch']))
g.add((uri, DCTERMS['created'], Literal(rfc3339(b.created),
datatype=XSD.dateTime)))
g.add((uri, DCTERMS['title'], Literal(b.name)))
g.add((uri, DCTERMS['creator'], abstract_uri(b.awardee, 'awardee')))
g.add((uri, NDNP['bag'], URIRef('/data/' + b.bag_relative_path)))
for issue in b.issues.all():
g.add((uri, ORE['aggregates'], abstract_uri(issue, 'issue')))
g.add((uri, ORE['isDescribedBy'], URIRef(rdf_uri(b))))
return g
def awardee_to_graph(a):
g = make_graph()
uri = abstract_uri(a, 'awardee')
g.add((uri, RDF.type, NDNP['Awardee']))
g.add((uri, DCTERMS['title'], Literal(a.name)))
for batch in a.batches.all():
g.add((abstract_uri(batch, 'batch'), DCTERMS['creator'], uri))
g.add((uri, RDFS.isDefinedBy, rdf_uri(a)))
return g
def make_graph():
g = ConjunctiveGraph()
g.bind('dc', DC)
g.bind('ore', ORE)
g.bind('owl', OWL)
g.bind('rda', RDA)
g.bind('bibo', BIBO)
g.bind('foaf', FOAF)
g.bind('ndnp', NDNP)
g.bind('dcterms', DCTERMS)
return g
def abstract_uri(m, fragment):
return URIRef(m.url.rstrip('/') + '#' + fragment)
def rdf_uri(m):
return URIRef(m.url.rstrip('/') + '.rdf')
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.