Script for replacing a list of broken links in both statements and references on Wikidata

fixBrokenLinks.py

import pywikibot
import pywikibot.data.wikidataquery as wdquery
import json
import codecs
BAD = json.load(codecs.open('fixLinks.json', 'r', 'utf-8'))
PATTERN = '2048005/Athena_Plus_ProvidedCHO_Nationalmuseum__Sweden__Inv__Nr__'

# add string replace for server change
BAD['//emp-web-22.zetcom.ch/eMuseumPlus?'] = \
    '//collection.nationalmuseum.se/eMuseumPlus?'


def findBad(claims, item, sourcedClaim=None):
    for p, pClaims in claims.iteritems():
        for pClaim in pClaims:
            pClaimTarget = pClaim.getTarget()
            if isinstance(pClaimTarget, (str, unicode)):
                if PATTERN in pClaimTarget:
                    print item, pClaimTarget
            if sourcedClaim is None:
                pSources = pClaim.getSources()
                for s in pSources:
                    findBad(s, item, sourcedClaim=pClaim)


def changeBad(item):
    for claims in item.claims.values():
        for claim in claims:
            claimTarget = claim.getTarget()
            if isinstance(claimTarget, (str, unicode)):
                for k, v in BAD.iteritems():
                    if k in claimTarget:
                        pywikibot.output('changing value: %s on %s' % (
                            claim.id, item.title()))
                        newTarget = claimTarget.replace(k, v)
                        claim.changeTarget(newTarget)
            # check in sources
            sources = claim.getSources()
            for source in sources:
                changeBadSource(source, claim)


def changeBadSource(source, sourcedClaim):
    foundBad = False
    for claim in source.values():
        if len(claim) != 1:
            print 'this should not happen'
            exit(1)
        claim = claim[0]
        claimTarget = claim.getTarget()
        if not isinstance(claimTarget, (str, unicode)):
            continue
        for k, v in BAD.iteritems():
            if k in claimTarget:
                foundBad = True
                newTarget = claimTarget.replace(k, v)
                claim.setTarget(newTarget)
    if foundBad:
        # must convert the OrderedDict format to a list of values
        claims = []
        for claim in source.values():
            claims.append(claim[0])
        # change source and update revid
        pywikibot.output('changing source: %s on %s' % (
            sourcedClaim.id, sourcedClaim.on_item.title()))
        data = sourcedClaim.repo.editSource(sourcedClaim, claims, new=False)
        sourcedClaim.on_item.latest_revision_id = data['pageinfo']['lastrevid']


def askWDQ():
    """
    Query Wikidata to fill the cache of entities which are actionable
    """
    query = u'CLAIM[195:842858]'
    wd_queryset = wdquery.QuerySet(query)

    wd_query = wdquery.WikidataQuery()
    data = wd_query.query(wd_queryset)

    if data.get('status').get('error') == 'OK':
        return data.get('items')
    return []


repo = pywikibot.Site().data_repository()
wdqResult = askWDQ()

# replace
for q in wdqResult:
    item = pywikibot.ItemPage(repo, 'Q%d' % q)
    item.exists()
    changeBad(item)

# check for any missed
for q in wdqResult:
    item = pywikibot.ItemPage(repo, 'Q%d' % q)
    item.exists()
    findBad(item.claims, item)