salgo60/PSIdatakommuner.py

## PSIdatakommuner.py
import sys
import urllib3
from SPARQLWrapper import SPARQLWrapper, JSON

endpoint_url = "https://query.wikidata.org/sparql"
http = urllib3.PoolManager()

#
# SPARQL https://w.wiki/bJV
query = """SELECT DISTINCT ?item ?wdLabel ?url ?PSIdata WHERE {
  ?item wdt:P31 wd:Q127448.
  OPTIONAL { ?item wdt:P856 ?url. }
  OPTIONAL { ?item wdt:P8402 ?openDataPortal. }
  OPTIONAL { ?item (wdt:P8402/wdt:P856) ?PSIdata. }
  OPTIONAL {
    ?item rdfs:label ?wdLabel.
    FILTER((LANG(?wdLabel)) = "sv")
  }
}
ORDER BY (?wdLabel) """


def get_results(endpoint_url, query):
    user_agent = "user Salgo50/%s.%s" % (sys.version_info[0], sys.version_info[1])
    sparql = SPARQLWrapper(endpoint_url, agent=user_agent)
    sparql.setQuery(query)
    sparql.setReturnFormat(JSON)
    return sparql.query().convert()


''' Check if URLS is ok'''
def check(url):
    try:
        r = http.request('GET', url)

        #print("\tStatus: ", r.status, " \t", url, " \t",print(r.headers))
        try:
            print("\t\tX-TEC-API-ROOT: ",r.headers['X-TEC-API-ROOT'])
        except:
            pass

        if r.status == 404:
            print("Status: ",r.status, " \t", url)
            return False
        return True
    except:
        pass
    return False

def getUrl(inUrl):
    if inUrl.endswith("/"):
        return inUrl
    else:
        return inUrl+"/"
    pass


def checkResult(results):
    global ok, notok , listSites
    for result in results["results"]["bindings"]:
        #print(result)
        currSite = {}
        currSite["URL"] = getUrl(result["url"]["value"]) + "PSIdata"
        currSite["item"] = result["item"]["value"].replace("http://www.wikidata.org/entity/","")
        currSite["name"] = result["wdLabel"]["value"]
        currSite["URLStatus"] = check(currSite["URL"])
        try:
            currSite["PSIdata"] = result["PSIdata"]["value"]
            currSite["PSIdataStatus"] = check(result["PSIdata"]["value"])
        except:
            pass
        listSites.append(currSite)
    return

listSites = []
results = get_results(endpoint_url, query)
checkResult(results)


def checkOKPSIData(listSites):
    ok = 0
    notok = 0
    for site in listSites:
        if site["URLStatus"]:
            ok += 1
        else:
            notok += 1
    return ok, notok


ok, notok  = checkOKPSIData(listSites)

print("OK: ",ok,"\t not ok",notok, "Antal: ", len(listSites))
	import sys
	import urllib3
	from SPARQLWrapper import SPARQLWrapper, JSON

	endpoint_url = "https://query.wikidata.org/sparql"
	http = urllib3.PoolManager()

	#
	# SPARQL https://w.wiki/bJV
	query = """SELECT DISTINCT ?item ?wdLabel ?url ?PSIdata WHERE {
	?item wdt:P31 wd:Q127448.
	OPTIONAL { ?item wdt:P856 ?url. }
	OPTIONAL { ?item wdt:P8402 ?openDataPortal. }
	OPTIONAL { ?item (wdt:P8402/wdt:P856) ?PSIdata. }
	OPTIONAL {
	?item rdfs:label ?wdLabel.
	FILTER((LANG(?wdLabel)) = "sv")
	}
	}
	ORDER BY (?wdLabel) """


	def get_results(endpoint_url, query):
	user_agent = "user Salgo50/%s.%s" % (sys.version_info[0], sys.version_info[1])
	sparql = SPARQLWrapper(endpoint_url, agent=user_agent)
	sparql.setQuery(query)
	sparql.setReturnFormat(JSON)
	return sparql.query().convert()


	''' Check if URLS is ok'''
	def check(url):
	try:
	r = http.request('GET', url)

	#print("\tStatus: ", r.status, " \t", url, " \t",print(r.headers))
	try:
	print("\t\tX-TEC-API-ROOT: ",r.headers['X-TEC-API-ROOT'])
	except:
	pass

	if r.status == 404:
	print("Status: ",r.status, " \t", url)
	return False
	return True
	except:
	pass
	return False

	def getUrl(inUrl):
	if inUrl.endswith("/"):
	return inUrl
	else:
	return inUrl+"/"
	pass


	def checkResult(results):
	global ok, notok , listSites
	for result in results["results"]["bindings"]:
	#print(result)
	currSite = {}
	currSite["URL"] = getUrl(result["url"]["value"]) + "PSIdata"
	currSite["item"] = result["item"]["value"].replace("http://www.wikidata.org/entity/","")
	currSite["name"] = result["wdLabel"]["value"]
	currSite["URLStatus"] = check(currSite["URL"])
	try:
	currSite["PSIdata"] = result["PSIdata"]["value"]
	currSite["PSIdataStatus"] = check(result["PSIdata"]["value"])
	except:
	pass
	listSites.append(currSite)
	return

	listSites = []
	results = get_results(endpoint_url, query)
	checkResult(results)


	def checkOKPSIData(listSites):
	ok = 0
	notok = 0
	for site in listSites:
	if site["URLStatus"]:
	ok += 1
	else:
	notok += 1
	return ok, notok


	ok, notok = checkOKPSIData(listSites)

	print("OK: ",ok,"\t not ok",notok, "Antal: ", len(listSites))