トピック投票の自動セルフ投票

:information_source: 概要 ユーザーがトピックを作成した際に、自動的にトピックに投票を有効にする
:hammer_and_wrench: リポジトリ GitHub - dereklputnam/discourse-topic-voting-auto-self-vote
:question: インストールガイド テーマまたはテーマコンポーネントのインストール方法
:open_book: Discourseテーマは初めてですか? Discourseテーマの初心者向けガイド

このテーマコンポーネントをインストール

コミュニティで投票権がMetaのように制限されている場合、OP(元の投稿者)の投票が自動的に行われないようにするのは理にかなっています(こちらこちらの議論を参照)。しかし、投票が無制限のコミュニティでは、これは問題となり、ユーザーは自分のトピックに投票することさえ忘れる可能性があります!したがって、このコンポーネントが必要になります。

設定

  • 自動投票カテゴリ: サイト全体で有効にしたくない場合は、適用対象のカテゴリを指定します。
  • 除外グループ: 特定のグループ(おそらく内部チーム)に適用したくない場合は、ここに追加します。
  • 訪問時の自動投票: OPが投票しなかった既存のトピックをクリーンアップするための簡単な方法です。

トピックのバックフィル

トピックの特定

OPが投票しなかったトピックを特定するには、このデータエクスプローラークエリを使用します。

-- [params]
-- null category_id :category_id
-- null string :category_slug
-- boolean :exclude_about = true

SELECT
  t.id AS topic_id,
  u.username AS "username",
  t.title AS "Topic Title",
  t.user_id AS "Author",
  t.created_at AS "Created At",
  c.name AS "Category Name",
  c.slug AS "Category Slug"
FROM topics t
JOIN users u ON u.id = t.user_id
JOIN categories c ON c.id = t.category_id
LEFT JOIN topic_voting_topic_vote_count tvvc ON tvvc.topic_id = t.id
LEFT JOIN topic_voting_votes tvv ON tvv.topic_id = t.id AND tvv.user_id = t.user_id
WHERE t.deleted_at IS NULL
  AND (
    tvvc.votes_count IS NULL OR tvv.id IS NULL
  )
  AND (
    :exclude_about = false
    OR t.title ILIKE 'about the % category' = false
  )
  AND (
    :category_id IS NULL OR c.id = :category_id
  )
  AND (
    :category_slug IS NULL OR c.slug = :category_slug
  )

ヒント: トピックを更新した後、このクエリを再実行して投票が投じられたことを確認してください :ballot_box:

APIによるバックフィル

次に、そのリストを使用してAPIスクリプトを実行し、投票をバックフィルします! 以下のスクリプトを使用するには、最初の2列以外の列を切り取る必要があります。 コミュニティ全体でトピックを識別しやすくするために、データエクスプローラークエリに含めました。

注意: 各ユーザーをなりすますため、APIキーはすべてのユーザーにスコープ設定されている必要があります。

使用したスクリプトを添付する権限がないため、ここにテキストとして示します。

バックフィルスクリプト
#!/usr/bin/env python3
"""
Discourse APIによるセルフ投票のバックフィル

このスクリプトは、自分のトピックに投票していないトピック作成者のために投票を行います。
Discourse APIを使用してユーザーになりすまし、彼らの代わりに投票を行います。

要件:
- Python 3.7以上
- requestsライブラリ (pip install requests)
- 代理実行権限を持つ管理者APIキー

使用方法:
1. 以下の設定セクションを更新します
2. topic_idとusernameの列を持つCSVファイルを用意します
3. 実行: python backfill_votes_api.py

CSV形式:
    topic_id,username
    12345,john_doe
    12346,jane_smith
"""

import csv
import time
import requests
from datetime import datetime

#==============================================================================
# 設定
#==============================================================================

# DiscourseインスタンスのURL(末尾のスラッシュなし)
DISCOURSE_URL = 'https://yourcommunity.com'

# 管理者APIキー(代理実行権限が必要)
API_KEY = 'YOUR_API_KEY_HERE'

# APIキーを所有する管理者ユーザー名
API_USERNAME = 'system'

# topic_idとusernameの列を持つCSVファイルへのパス
CSV_FILE = 'topics_to_vote.csv'

# ドライランモード - 実際に投票する場合はFalseに設定
DRY_RUN = True

# レート制限を避けるためのAPI呼び出し間の遅延(秒)
DELAY_BETWEEN_REQUESTS = 0.5

#==============================================================================
# スクリプト
#==============================================================================

def cast_vote(topic_id: int, username: str) -> dict:
    """
    特定のユーザーとしてトピックに投票を行います。

    Args:
        topic_id: 投票するトピックのID
        username: 投票時に代理実行するユーザー名

    Returns:
        'success'ブール値と'message'文字列を含む辞書
    """
    url = f"{DISCOURSE_URL}/voting/vote"

    headers = {
        'Api-Key': API_KEY,
        'Api-Username': username,  # ユーザーを代理実行
        'Content-Type': 'application/json'
    }

    data = {
        'topic_id': topic_id
    }

    try:
        response = requests.post(url, headers=headers, json=data)

        if response.status_code == 200:
            return {'success': True, 'message': 'Vote cast successfully'}
        elif response.status_code == 422:
            # 通常、すでに投票済みまたは投票が無効な場合
            return {'success': False, 'message': 'Already voted or voting not enabled'}
        elif response.status_code == 403:
            return {'success': False, 'message': 'Permission denied - check API key permissions'}
        elif response.status_code == 404:
            return {'success': False, 'message': 'Topic not found or voting not enabled on category'}
        else:
            return {'success': False, 'message': f'HTTP {response.status_code}: {response.text[:200]}'}

    except requests.RequestException as e:
        return {'success': False, 'message': f'Request error: {str(e)}'}

def main():
    print("\n" + "=" * 60)
    print("Discourse APIによるセルフ投票のバックフィル")
    print("=" * 60)
    print(f"モード: {'ドライラン(変更なし)' if DRY_RUN else 'ライブ(投票が実行されます)'}")
    print(f"ターゲット: {DISCOURSE_URL}")
    print(f"CSVファイル: {CSV_FILE}")
    print("=" * 60 + "\n")

    # CSVファイルを読み込む
    try:
        with open(CSV_FILE, 'r', newline='', encoding='utf-8') as f:
            reader = csv.DictReader(f)
            rows = list(reader)
    except FileNotFoundError:
        print(f"エラー: CSVファイルが見つかりません: {CSV_FILE}")
        print("\n以下の形式でCSVファイルを作成してください:")
        print("topic_id,username")
        print("12345,john_doe")
        print("12346,jane_smith")
        return
    except Exception as e:
        print(f"エラー: CSVファイルの読み込みに失敗しました: {e}")
        return

    if not rows:
        print("エラー: CSVファイルが空です")
        return

    # CSV列の検証
    columns = set(rows[0].keys())

    if 'topic_id' not in columns or 'username' not in columns:
        print(f"エラー: CSVには topic_id と username の列が必要です")
        print(f"見つかった列: {columns}")
        return

    print(f"{len(rows)}件のトピックを処理します\n")

    # 各行を処理
    success_count = 0
    skip_count = 0
    error_count = 0

    for i, row in enumerate(rows, 1):
        topic_id = row['topic_id'].strip()
        username = row['username'].strip()

        print(f"[{i}/{len(rows)}] トピック #{topic_id} @{username}", end=" ")

        if DRY_RUN:
            print("-> 投票予定")
            success_count += 1
        else:
            result = cast_vote(int(topic_id), username)

            if result['success']:
                print("-> 投票済み!")
                success_count += 1
            elif 'Already voted' in result['message']:
                print("-> すでに投票済み(スキップ)")
                skip_count += 1
            else:
                print(f"-> エラー: {result['message']}")
                error_count += 1

            # レート制限
            if i < len(rows):
                time.sleep(DELAY_BETWEEN_REQUESTS)

    # サマリー
    print("\n" + "=" * 60)
    print("サマリー")
    print("=" * 60)
    print(f"合計トピック数: {len(rows)}")
    print(f"投票{'実行予定' if DRY_RUN else '実行済み'}: {success_count}")
    print(f"すでに投票済み(スキップ): {skip_count}")
    print(f"エラー: {error_count}")

    if DRY_RUN:
        print("\n** ドライラン完了 **")
        print("投票を実行するには、DRY_RUN = False に設定して再度実行してください。")

    print("")


if __name__ == '__main__':
    main()
「いいね!」 4