利用者:Hatukanezumi/仮リンクの整理/aggregateTentativeLinks.py

1. -*- python -*-
2. -*- coding: utf-8 -*-

"""

aggregateTentativeLinks.pyは、{{仮リンク}}テンプレートの使用状況を調査し、結果を特定のページ (複数) に投稿するボットです。通常ボットとしてイメージされるプログラムとは異なり、このボットは大量のページからデータを取得しながら、あらかじめ決められたごくわずかのページしか変更しません。

必要なソフトウェア

• pywikipedia。2010年11月ころのtrunkでテストしていますが、最近のバージョンならたいてい大丈夫だと思います。
• Pythonインタプリタ。pywikipediaが動くバージョンのもの。

手順

1. pywikipediaを、自分のボット用アカウントでログインできるように設定します。
2. 当ページのソースをダウンロードして保存します (画面をコピー・ペーストしてもうまく動かないかもしれません)。保存する際の文字コードはUTF-8、改行はpywikipediaをインストールしたオペレーティングシステムの改行コードにします。
3. 保存したファイルの名前を「aggregateTentativeLinks.py」にして、pywikipediaのディレクトリに複写します。

1. 下記「コード」の「基本設定」の箇所を適当に修正します。
2. OUTPUTDIRで設定したディレクトリがなければ、作ります。

aggregateTentativeLinks.pyは、つぎの二段階に分けて実行できます。

情報を取得して解析し、OUTPUTDIR下に保存する。

python aggregateTentativeLinks.py -retrieve

保存した情報を投稿する。

python aggregateTentativeLinks.py -put

「-retrieve」と「-put」のいずれかは、かならず指定する必要があります。両方指定すると、情報の取得・解析と投稿を続けて実行します。

ほかのオプション。

-max:数

-retrieveの場合、仮リンクのあるページのうち、指定した数だけ処理します。数を制限するだけで、どのページを処理するかは選べません。

-comment:テキスト

-put の場合、投稿時の要約欄の内容。

-always

-put の場合、投稿するかどうかを確認せずに実行する。

まず、オプションに「-retrieve -max:小さな数」を指定して、このボットがどんなふうに情報を収集するかを見てください。つぎに「-put」を指定すれば、収集した情報がどのように投稿されるかがわかります。本格的に運用するには、「-max:数」オプションを指定せずに動かします。完全に自動化してもよいと思ったらはじめて、「-always」オプションを追加して投稿します。

retrieve処理では、処理の途中結果を外部記憶などに保存しません。そのため、なんらかの原因で実行が中断すると、最初からやりなおしです。処理にかかる時間の大半はメディアウィキサーバとの通信が占めるので、実行するコンピュータの性能はあまり関係ありません。

• 仮リンクテンプレートを使ったページ約2500に対して、実測で3-5時間程度かかりました。
• メモリは、AMD64 Linux上のpythonでおよそ200MB必要でした。

pywikipediaの現時点での制限により、仮リンクテンプレートを使っているページが5000を超えると、すべての項目の情報を取得できなくなります (と思います)。

aggregateTentativeLinks.pyは、ウィキペディアの記事と同じライセンスにしたがって配布、利用、変更、再配布、二次著作物の作成等を行えます。

オリジナルの版は、このページのこの版です。

"""
###
### 基本設定。LANG、FAMILY、TEMPLATENAMEは通常は変更不要。
###
LANG = 'ja' # 対象プロジェクトの言語
FAMILY = None # プロジェクトファミリ (Noneならuser-config.py
 # の設定にしたがう)
TEMPLATENAME = '仮リンク' # 仮リンクテンプレートのページ名 (名前空間なし)
LISTPAGES = [
 # 報告先ページの名前空間番号とページ名のプリフィクス。
 # これらで始まる名前のすべてのページから{{jareq}}テンプレートを抽出する。
 (4, '多数の言語版にあるが日本語版にない記事'),
]
OUTPUTDIR = '/var/tmp/wiki' # 結果を出力するディレクトリ。存在すること。
 # 結果を投稿する先のメインページ。
 # 複数のサブページに投稿する。
OUTPUTPAGEBASE = '利用者:Hatukanezumi/仮リンクの整理'
###
### ここから後は変更の必要はありません。
###
import os
import sys
import re
from wikipedia import Site, Page, handleArgs, inputChoice, output, stopme
#from catlib import Category
SITE = Site(LANG, FAMILY)
TEMPLATENAME = Page(SITE, 'Template:'+unicode(TEMPLATENAME, 'utf-8')).titleWithoutNamespace()
class ProposedArticles:
 """
 解析結果を保持するためのクラス。クラスにした意味があまりない。
 """
 def __init__(self):
 self.hint = {}
 self.ref = {}
 #self.cat = {}
 self.pages = []
 def addHint(self, proposed, project, pagename):
 p = self.hint.get(proposed, set())
 p.update([Page(Site(project, FAMILY), pagename).aslink().replace('[[', '').replace(']]', '')])
 self.hint[proposed] = p
 #def addCat(self, proposed, category):
 # p = self.cat.get(proposed, {})
 # p[category] = p.get(category, 0) + 1
 # self.cat[proposed] = p
 def addRef(self, proposed, referer):
 p = self.ref.get(proposed, set())
 p.update([referer.title()])
 self.ref[proposed] = p
def getListedPages():
 """
 WP:JAREQから、報告ずみの項目名を取得
 """
 listedPages = {}
 for ns, pfx in LISTPAGES:
 for page in SITE.prefixindex(unicode(pfx, 'utf-8'), ns, False):
 output('Getting: ' + page.aslink().encode('utf-8'))
 for tname, args in page.templatesWithParams(get_redirect=True):
 if tname.lower() <> 'jareq':
 continue
 alt = [x[4:] for x in args if x.startswith('alt=')]
 args = [x for x in args if x.find('=') < 0]
 reqPage = None
 try:
 reqPage = Page(SITE, args[1])
 except:
 continue
 p = listedPages.get(reqPage.title(), set())
 p.update([reqPage.title()])
 listedPages[reqPage.title()] = p
 for al in alt:
 for a in re.split(r'(?<=\]\])/|/(?=\[\[)', al):
 if re.match(r'\[\[.+\]\]', a) and \
 not re.match(r'.+\]\].+', a):
 try:
 a = Page(SITE, a.replace('[[','').replace(']]',''))
 except:
 continue
 p = listedPages.get(a.title(), set())
 p.update([reqPage.title()])
 listedPages[a.title()] = p
 output('listed: %d' % len(listedPages))
 return listedPages
def aggregate(proposedArticles, maxCount):
 """
 {{仮リンク}}の使用情報を取得する。
 同テンプレートを使用しているすべてのページからテンプレートのマークアップ
 を抽出し、推奨項目名、参考リンク情報を取得する。
 """
 templatePage = Page(SITE, 'Template:'+TEMPLATENAME)
 count = 0
 for page in templatePage.getReferences(follow_redirects=False,
 onlyTemplateInclusion=True):
 # 標準名前空間のページのみを走査する
 if page.namespace() <> 0:
 continue
 # DEBUG
 output('Analyzing ' + page.title().encode('utf-8'))
 ## 呼び出し元ページからカテゴリを取得する
 #cats = [c.titleWithoutNamespace() for c in page.categories()]
 # テンプレートを処理する
 for tname, args in page.templatesWithParams(get_redirect=True):
 if tname <> TEMPLATENAME:
 continue
 args = [arg.strip() for arg in args
 if not arg.strip().startswith('label=')]
 if not len(args): # 引数が必要
 continue
 try:
 proposed = args[0]
 args = args[1:]
 proposed = proposed.split('{{!}}')[0] # 誤用への対応
 if not proposed.strip():
 raise
 proposed = Page(SITE, proposed).title()
 except:
 output('Bad name of proposed article: %r' % proposed)
 continue
 # 呼び出し元ページ
 proposedArticles.addRef(proposed, page)
 ## 呼び出し元ページのカテゴリを仮項目名に対応づける
 #for cat in cats:
 # proposedArticles.addCat(proposed, cat)
 # 参考リンクを取得する
 try:
 while len(args):
 proposedArticles.addHint(proposed, args[0], args[1])
 args = args[2:]
 except:
 output('Bad args: %r' % args)
 continue
 count += 1
 if 0 < maxCount and maxCount <= count:
 break
def dump(proposedArticles, listedPages):
 """
 取得した情報を整理して、ファイルに出力する。
 
 * listed.wiki - JAREQ掲載ずみ
 * redirect.wiki - ページは存在するがリダイレクト
 * disambig.wiki - 曖昧さ回避ページとして存在する
 * empty.wiki - 存在するが内容がない
 * exists.wiki - 以上以外で立項ずみ
 * synonym.wiki - 未立項だが、言語間リンク中にホームウィキの項目がある
 * unknown.wiki - 未立項。言語間リンクが取得できない
 * 1.wiki, 2.wiki, ... - 以上のどれでもない。未立項。
 参考リンクからたどれる言語間リンクの数により分類
 """
 outputs = {}
 for proposed in proposedArticles.ref.keys():
 page = Page(SITE, proposed)
 # 分類する
 g = 'unknown'
 synonyms = []
 if listedPages.has_key(page.title()):
 # WP:JAREQに報告ずみのもの。別名があればそれも追加
 g = 'listed'
 synonyms = [Page(SITE, x) for x in listedPages[page.title()]
 if x <> page.title()]
 elif page.exists():
 # ページが存在する場合。リダイレクト、曖昧さ回避、白紙は分ける
 if page.isRedirectPage():
 g = 'redirect'
 elif page.isDisambig():
 g = 'disambig'
 elif page.isEmpty():
 g = 'empty'
 else:
 g = 'exists'
 else:
 # 参考リンクのページから言語間リンクを抽出
 interwiki = set()
 for hint in proposedArticles.hint.get(proposed, set()):
 try:
 hintPage = Page(SITE, hint)
 if hintPage.isRedirectPage():
 hintPage = hintPage.getRedirectTarget()
 interwiki.update([x.aslink().replace('[[','').replace(']]','') for x in hintPage.interwiki()])
 interwiki.update([hintPage.aslink().replace('[[','').replace(']]','')])
 except:
 output('Failed to get interwiki: %r' % hint)
 # 言語間リンクにホームウィキの項目があればシノニムとして抽出
 synonyms = [Page(SITE, p) for p in interwiki
 if Page(SITE, p).site().language() == LANG]
 # シノニムがないものは言語間リンク数で分類
 if len(synonyms):
 g = 'synonym'
 elif len(interwiki):
 g = len(interwiki)
 out = outputs.get(g, [])
 # DEBUG
 output('Dump: %s: %s' % (g, proposed.encode('utf-8')))
 # 整形する
 o = ['[[:%s|%s]]' % (h, h.split(':')[0]) for h in proposedArticles.hint.get(proposed, set())]
 o.sort()
 hints = '/'.join(o)
 o = ['[[%s]]' % r for r in proposedArticles.ref.get(proposed, set())]
 o.sort()
 refs = '/'.join(o)
 o = [s.aslink() for s in synonyms]
 o.sort()
 syns = '/'.join(o)
 f = (page.aslink().encode('utf-8'),
 hints.encode('utf-8'),
 refs.encode('utf-8'),
 page.aslink().replace('[[', '[[special:whatLinksHere/').replace(']]', '|...]]').encode('utf-8'))
 if g == 'synonym' or len(synonyms):
 f += (syns.encode('utf-8'),)
 out.append('* %s<small>(%s)</small> ←%s%s<br/>≈%s' % f)
 else:
 out.append('* %s<small>(%s)</small> ←%s%s' % f)
 outputs[g] = out
 # 以前のファイルを消す
 for path in os.listdir(OUTPUTDIR):
 if not path.endswith('.wiki'):
 continue
 try:
 os.unlink(os.path.join(OUTPUTDIR, path))
 except:
 output('Failed to remove: %s' % path)
 # ファイルを出力する
 for k, out in outputs.items():
 fp = open(os.path.join(OUTPUTDIR, '%s.wiki' % k), 'w')
 out.sort()
 print >>fp, "\n".join(out),
 fp.close()
def put(pagename, commentText, data, always):
 count = 0
 comment = commentText
 text = "__TOC__\n"
 for filename, title in data:
 path = os.path.join(OUTPUTDIR, filename)
 if os.path.exists(path):
 lines = [l for l in file(path)]
 if commentText is None:
 if not comment:
 comment = ''
 else:
 comment += '; '
 comment += unicode('%s%d件' % (title, len(lines)), 'utf-8')
 count += len(lines)
 text += "== %s ==\n%s\n\n" % (title, ''.join(lines))
 comment = unicode('%d件: %s', 'utf-8') % (count, comment)
 if 200 < len(comment) or 250 <= len(comment.encode('utf-8')):
 comment = unicode(comment[:197].encode('utf-8')[:246], 'utf-8', 'ignore') + u'...'
 page = Page(SITE, unicode(pagename, 'utf-8'))
 if always:
 choice = 'y'
 else:
 output(comment)
 choice = inputChoice(
 'Do you update %s' % page.aslink(),
 ['Yes', 'No', 'Quit'],
 ['y', 'N', 'q'], 'N')
 if choice == 'q':
 sys.exit(0)
 elif choice == 'y':
 page.put(unicode(text, 'utf-8'), comment)
 else:
 return
def main(*argv):
 toDo = {}
 maxCount = 0
 commentText = None
 always = False
 for arg in handleArgs(*argv):
 if arg == '-retrieve':
 toDo['retrieve'] = True
 elif arg == '-put':
 toDo['put'] = True
 elif arg.startswith('-max:'):
 try:
 maxCount = int(arg[5:])
 except:
 output('Illegal argument: %s' % arg)
 sys.exit(1)
 elif arg.startswith('-comment:'):
 commentText = arg[9:]
 elif arg == '-always':
 always = True
 else:
 output('Unknown argument: %s' % arg)
 sys.exit(1)
 if not toDo.has_key('retrieve') and not toDo.has_key('put'):
 output('At least either of -retrieve and -put is required.')
 sys.exit(1)
 if toDo.has_key('retrieve'):
 proposedArticles = ProposedArticles()
 aggregate(proposedArticles, maxCount)
 listedPages = getListedPages()
 dump(proposedArticles, listedPages)
 if toDo.has_key('put'):
 put(OUTPUTPAGEBASE + '/要検討',
 commentText,
 [('unknown.wiki', 'プロジェクト数不明'),
 ('disambig.wiki', '曖昧さ回避ページ'),
 ('redirect.wiki', 'リダイレクト'),
 ('synonym.wiki', 'シノニム'),
 ('empty.wiki', '白紙')],
 always)
 put(OUTPUTPAGEBASE + '/立項・報告ずみ',
 commentText,
 [('exists.wiki', '立項ずみ'),
 ('listed.wiki', 'WP:JAREQに報告ずみ')],
 always)
 put(OUTPUTPAGEBASE + '/少数の言語版',
 commentText,
 [('%d.wiki' % x, '%d言語版' % x) for x in range(4, 0, -1)],
 always)
 put(OUTPUTPAGEBASE + '/10-5言語版',
 commentText,
 [('%d.wiki' % x, '%d言語版' % x) for x in range(10, 4, -1)],
 always)
 put(OUTPUTPAGEBASE + '/15-11言語版',
 commentText,
 [('%d.wiki' % x, '%d言語版' % x) for x in range(15, 10, -1)],
 always)
 put(OUTPUTPAGEBASE + '/20-16言語版',
 commentText,
 [('%d.wiki' % x, '%d言語版' % x) for x in range(20, 15, -1)],
 always)
 put(OUTPUTPAGEBASE + '/21言語版以上',
 commentText,
 [('%d.wiki' % x, '%d言語版' % x) for x in range(100, 20, -1)],
 always)
if __name__ == '__main__':
 try:
 main()
 except:
 raise
 #XXXstopme()