import tweepy import logging from typing import List, Dict, Any, Optional import time import requests logger = logging.getLogger(__name__) class XApiClient: """X (Twitter) API v2連携クライアント""" def __init__(self, settings: Dict[str, Any]): """ スプレッドシートから読み込んだ settings dict を受け取り、 Tweepy Client を初期化する。 token.json は使用しない。 """ try: def _val(key: str) -> str: """スプレッドシートの未入力プレースホルダーを空文字として扱うヘルパー""" v = settings.get(key, "") if not v or "値を入力" in str(v) or str(v).strip() == "—": return "" return str(v).strip() api_key = _val("api_key") api_key_secret = _val("api_key_secret") access_token = _val("access_token") access_token_secret = _val("access_token_secret") # Bearer Tokenは設定に直接書いてあればそれを使用。 # なければ api_key + api_key_secret から動的に取得する。 bearer_token = _val("bearer_token") if not bearer_token: auth_res = requests.post( "https://api.twitter.com/oauth2/token", auth=(api_key, api_key_secret), data={"grant_type": "client_credentials"} ) if auth_res.status_code == 200: bearer_token = auth_res.json().get("access_token", "") logger.info("Bearer Tokenをapi_key/api_key_secretから動的に取得しました。") else: logger.warning(f"Bearer Token動的取得に失敗しました({auth_res.status_code})。検索機能が使えない可能性があります。") self.client = tweepy.Client( bearer_token=bearer_token or None, consumer_key=api_key or None, consumer_secret=api_key_secret or None, access_token=access_token or None, access_token_secret=access_token_secret or None, wait_on_rate_limit=False # 自前で429ハンドリングするためFalse ) logger.info("X API クライアントの初期化に成功しました。") except Exception as e: logger.error(f"X APIクライアントの初期化に失敗しました: {e}") raise def search_tweets(self, settings: Dict[str, Any]) -> List[Dict[str, Any]]: """ 設定に基づいてツイートを検索する。 - キーワードローテーション: 全キーワードから毎回ランダムにN個を選択 - NGワードを含むものは除外 (-キーワード) - URL付きツイートを除外 (-has:links またはコード側で除外) - is:retweet, is:reply の除外 """ import random # --- クエリ構築 --- raw_keywords = settings.get("search_keywords", "") keywords = [k.strip() for k in raw_keywords.split(",") if k.strip()] if not keywords: logger.warning("検索キーワードが設定されていません。") return [] # キーワードローテーション: プールからランダムにN個選択 # (スプレッドシートの入力が空や不正な場合へのガード付き) try: rot_val = settings.get("keyword_rotation_count") rotation_count = int(rot_val) if rot_val and str(rot_val).strip() else len(keywords) except (ValueError, TypeError): rotation_count = len(keywords) selected_keywords = random.sample(keywords, max(1, min(rotation_count, len(keywords)))) logger.info(f"キーワードローテーション: {len(keywords)}個中 {len(selected_keywords)}個を選択 → {selected_keywords}") # 基盤となる検索条件: ("A B" OR "C") -is:retweet -is:reply # 厳密なフレーズ一致を強制するため、常に引用符で囲む query_base = "(" + " OR ".join(f'"{k}"' for k in selected_keywords) + ")" query = f"{query_base} -is:retweet -is:reply" # URL除外設定 if settings.get("exclude_urls", True): query += " -has:links" # NGワード制限 raw_ng_words = settings.get("ng_words", "") ng_words = [ng.strip() for ng in raw_ng_words.split(",") if ng.strip()] for ng in ng_words: query += f' -"{ng}"' logger.info(f"API発行クエリ: {query}") max_results = max(10, int(settings.get("search_limit", 10))) # 最小10（API制限） try: # 検索上限を考慮（ここではAPIコールのリミット） # expansionsでユーザー情報も同時に取得する response = self.client.search_recent_tweets( query=query, max_results=max_results, expansions=['author_id'], user_fields=['name', 'username', 'description', 'url'], tweet_fields=['created_at', 'text'] ) if not response.data: logger.info("条件に一致するツイートが見つかりませんでした。") return [] # ユーザー情報のマッピング (author_id -> Userオブジェクト) users_map = {u.id: u for u in response.includes['users']} if 'users' in response.includes else {} results = [] for tweet in response.data: user = users_map.get(tweet.author_id) # さらなるコード側でのNGワードフィルタ（APIのクエリ漏れ対策） tweet_text = tweet.text if any(ng in tweet_text for ng in ng_words): continue username = f"@{user.username}" if user else "" # プロフィールにNGワードが含まれるかもチェック profile_desc = user.description if user and user.description else "" if any(ng in profile_desc for ng in ng_words): continue results.append({ "tweet_id": str(tweet.id), "tweet_text": tweet_text, "tweet_url": f"https://x.com/{user.username}/status/{tweet.id}" if user else "", "created_at": tweet.created_at.strftime("%Y/%m/%d %H:%M") if tweet.created_at else "", "author_id": str(tweet.author_id), "username": username, "profile": profile_desc, "profile_url": f"https://x.com/{user.username}" if user else "" }) logger.info(f"{len(results)} 件の有効なツイートを取得しました。") return results except tweepy.TooManyRequests: logger.error("APIのレート制限（429 Too Many Requests）またはUsage Capに到達しました。") return [] except Exception as e: logger.error(f"ツイート検索中にエラーが発生しました: {e}") return [] def get_user_id_by_username(self, username: str) -> Optional[str]: """ユーザー名（@〇〇）から数値のユーザーIDを取得する""" clean_username = username.lstrip('@') try: res = self.client.get_user(username=clean_username) if res.data: return str(res.data.id) return None except Exception as e: logger.error(f"ユーザーID取得エラー {username}: {e}") return None def like_tweet(self, tweet_id: str) -> str: """指定したツイートIDにいいねをする""" logger.info(f"ツイートID: {tweet_id} に「いいね」を実行します...") try: res = self.client.like(tweet_id=tweet_id) if res.data and res.data.get('liked'): return "成功" return "失敗(API応答異常)" except tweepy.TooManyRequests: logger.error("Quota Exceeded: 予算上限またはレート制限到達") return "エラー:制限到達" except tweepy.errors.Forbidden as e: logger.error(f"403 Forbidden (いいね実行) 詳細エラー: {e.response.text if hasattr(e, 'response') else str(e)}") return "エラー: 凍結/権限不足" except Exception as e: logger.error(f"いいね実行エラー {tweet_id}: {e}") return f"エラー: {e}" def post_tweet(self, text: str) -> str: """新規ツイートを投稿する""" logger.info(f"新規ポストを実行します: {text[:20]}...") try: res = self.client.create_tweet(text=text) if res.data and res.data.get('id'): return "成功" return "失敗(API応答異常)" except tweepy.TooManyRequests: logger.error("Quota Exceeded: 予算上限またはレート制限到達") return "エラー:制限到達" except tweepy.errors.Forbidden as e: logger.error(f"403 Forbidden (ポスト実行) 詳細エラー: {e.response.text if hasattr(e, 'response') else str(e)}") return "エラー: 凍結/権限不足" except Exception as e: logger.error(f"ポスト実行エラー: {e}") return f"エラー: {e}" def follow_user(self, user_id: str) -> str: """指定したユーザーIDをフォローする""" logger.info(f"ユーザーID: {user_id} に「フォロー」を実行します...") try: res = self.client.follow_user(target_user_id=user_id) if res.data and res.data.get('following') or res.data.get('pending_follow'): return "成功" # 既にフォローしている場合などは data 内のステータスが異なる場合がある return "成功(既フ含む)" except tweepy.TooManyRequests: logger.error("Quota Exceeded: 予算上限またはレート制限到達") return "エラー:制限到達" except tweepy.errors.Forbidden as e: logger.error(f"403 Forbidden (フォロー実行) 詳細エラー: {e.response.text if hasattr(e, 'response') else str(e)}") return "エラー: 凍結/権限不足" except Exception as e: logger.error(f"フォロー実行エラー {user_id}: {e}") return f"エラー: {e}" def reply_tweet(self, tweet_id: str, reply_text: str) -> str: """指定したツイートIDに対してリプライを送信する""" logger.info(f"ツイートID: {tweet_id} にリプライを実行します: {reply_text[:20]}...") try: res = self.client.create_tweet(text=reply_text, in_reply_to_tweet_id=tweet_id) if res.data and res.data.get('id'): return "成功" return "失敗(API応答異常)" except tweepy.TooManyRequests: logger.error("Quota Exceeded: 予算上限またはレート制限到達") return "エラー:制限到達" except tweepy.errors.Forbidden as e: err_msg = e.response.text if hasattr(e, 'response') else str(e) logger.error(f"403 Forbidden 詳細エラー: {err_msg}") # 返信制限（Conversation Control）によるエラーを判定 if "restricted" in err_msg or "not allowed" in err_msg: return "エラー: 返信制限あり" return "エラー: 凍結/権限不足" except Exception as e: logger.error(f"リプライ実行エラー {tweet_id}: {e}") return f"エラー: {e}" if __name__ == "__main__": # 動作確認用 logging.basicConfig(level=logging.INFO) try: from x_sheets_writer import XSheetsManager creds = r"C:\Windows\LOCALSTRAGE\GoogleAntiGravity\00_Common\gen-lang-client-0965294635-8bc2247ad0d2.json" sid = "1icqi5ZjqjRUOcGnxtDyHw1-gXj-osmGjB8twUnKB6i0" manager = XSheetsManager(creds, sid) settings = manager.get_settings() client = XApiClient(settings) # settings から認証情報を取得 tweets = client.search_tweets(settings) for t in tweets: print(f"- {t['username']} : {t['tweet_text'][:30]}... ({t['tweet_url']})") except Exception as test_e: print(f"Test failed: {test_e}")