import gspread import logging from typing import Dict, List, Any, Optional import os import logging from typing import Dict, List, Any, Optional logger = logging.getLogger(__name__) class XSheetsManager: """X自動集客ツール用 Googleスプレッドシート連携クラス""" # シート名定数 SHEET_ACCOUNT = "👥アカウント管理" SHEET_TWEET = "📋ツイート管理" SHEET_POST = "📨自ポスト管理" # === 列定義 (0始まりインデックス: A列=0, B列=1, C列=2 ...) === # 万が一スプレッドシートの列が変更された場合は、ここの数値を変更するだけで対応可能です。 # 【アカウント管理シート】 # ユーザー指定の通り: C=名, D=プロフ, E=URL, F=日時, G=リプライ対象, H=フォロー対象, J=フォロー結果 COL_ACC_USERNAME = 2 # C列 (2) COL_ACC_PROFILE = 3 # D列 (3) COL_ACC_URL = 4 # E列 (4) COL_ACC_TIMESTAMP = 5 # F列 (5) COL_ACC_FLG_REPLY = 6 # G列 (6) COL_ACC_FLG_FOLLOW = 7 # H列 (7) COL_ACC_RESULT = 9 # J列 (9) COL_ACC_FOLLOW_RECOMMENDED = 10 # K? (10) COL_ACC_ACCOUNT_SCORE = 11 # L? (11) COL_ACC_SCORE_REASON = 12 # M? (12) COL_ACC_EXCLUDE_REASON = 13 # N? (13) # 【ツイート管理シート】 # 想定: C=名, D=本文, E=URL, F=日時, G=アカステータス, H=リプライ対象, J=AIコメント案, L=いいね結果, M=リプライ結果 COL_TW_USERNAME = 2 # C列 (2) COL_TW_TEXT = 3 # D列 (3) COL_TW_URL = 4 # E列 (4) COL_TW_TIMESTAMP = 5 # F列 (5) COL_TW_ACC_STATUS = 6 # G列 (6) COL_TW_FLG_REPLY = 7 # H列 (7) COL_TW_AI_COMMENT = 9 # J列 (9) COL_TW_LIKE_RESULT = 11 # L列 (11) COL_TW_REPLY_RESULT = 12 # M列 (12) COL_TW_LIKE_RECOMMENDED = 13 # N? (13) COL_TW_FOLLOW_RECOMMENDED = 14 # O? (14) COL_TW_POST_SCORE = 15 # P? (15) COL_TW_ACCOUNT_SCORE = 16 # Q? (16) COL_TW_SCORE_REASON = 17 # R? (17) COL_TW_EXCLUDE_REASON = 18 # S? (18) # 【自ポスト管理シート】 # C=キーワード, D=場面, E=補足, F=投稿文, G=有効FLG, H=実行結果 COL_POST_KEYWORDS = 2 # C列 (2) COL_POST_SCENE = 3 # D列 (3) COL_POST_NOTE = 4 # E列 (4) COL_POST_TEXT = 5 # F列 (5) COL_POST_ENABLE = 6 # G列 (6) COL_POST_RESULT = 7 # H列 (7) def __init__(self, creds_file: str, spreadsheet_id: str): """初期化。gspreadでスプレッドシートを開く""" try: self.gc = gspread.service_account(filename=creds_file) self.sh = self.gc.open_by_key(spreadsheet_id) self._account_insert_row_cache = None self._account_data_start_cache = None self._tweet_insert_row_cache = None self._tweet_data_start_cache = None logger.info(f"スプレッドシート '{self.sh.title}' に接続しました。") except Exception as e: logger.error(f"スプレッドシートへの接続に失敗しました: {e}") raise def _detect_settings_end_row(self, ws) -> int: """ C列（設定項目名の列）の2行目以降をスキャンし、最初に空白になった行番号を返す。 設定エリアの終端行を動的に検出するためのヘルパー。 - C1 = ヘッダーまたは空（スキップ） - C2〜 = 設定項目名 - 最初の空白 = 設定エリアの終端 """ c_col = ws.col_values(3) # C列（1始まりで3番目）の全値を取得 # 2行目(インデックス1)から探索 for i in range(1, len(c_col)): if not c_col[i].strip(): return i + 1 # 1始まりの行番号で返す（例: C12が空 → 12を返す） # 空白行が見つからない場合はシートの末尾+1 return len(c_col) + 1 def _detect_data_start_row(self, ws) -> int: """ データ开始行を動的に検出する。 ロジック: 1) C列の設定エリア終端（最初の空白行）を検出 2) 空白行の次に最初に内容のある行 = ヘッダー行 3) ヘッダー行の次の行 = データ開始行 """ c_col = ws.col_values(3) # ステップ1: 設定エリア終端（空白行の行番号）を導く settings_end = self._detect_settings_end_row(ws) # 1始まり行番号 # ステップ2: 空白行以降で最初に内容のある行 = ヘッダー行 header_row = settings_end # デフォルトフォールバック for i in range(settings_end - 1, len(c_col)): if c_col[i].strip(): header_row = i + 1 # 1始まり行番号 break # ステップ3: データ开始行 = ヘッダー行 + 1 data_start = header_row + 1 logger.debug(f"データ開始行を検出: 設定終端={settings_end}行, ヘッダー={header_row}行, データ開始={data_start}行") return data_start def get_settings(self) -> Dict[str, Any]: """ 各シートの上部（設定領域）から共通設定を読み込む。 設定領域の下端は、A列の空白セルが最初に現れる行で自動判定する。 """ settings = {} def _parse_settings_from_sheet(ws): """ シートを受け取り、設定エリアを動的に検出してkey-valueを返す。 - C列: 設定名（人が読む用ラベル。コードからは参照しない） - D列: 変数名（コードが参照するキー） - E列: 値 """ end_row = self._detect_settings_end_row(ws) if end_row <= 2: return {} # 2行目がすでに空白 = 設定なし # C列〜E列を2行目〜終端の1行前まで取得 range_str = f"C2:E{end_row - 1}" logger.info(f"設定読み込みレンジ: {ws.title}!{range_str}") result = {} rows = ws.batch_get([range_str])[0] for row in rows: # D列（変数名）が必須。E列（値）が未入力の場合は空文字扱い if len(row) >= 2 and row[1].strip(): key = row[1].strip() # D列 = 変数名 val = row[2].strip() if len(row) >= 3 else "" # E列 = 値 if val.upper() == "TRUE": val = True elif val.upper() == "FALSE": val = False result[key] = val return result # 👥アカウント管理シートからの設定読み込み try: ws_account = self.sh.worksheet(self.SHEET_ACCOUNT) settings.update(_parse_settings_from_sheet(ws_account)) except Exception as e: logger.warning(f"'{self.SHEET_ACCOUNT}' の設定読み込み中にエラー: {e}") # 📋ツイート管理シートからの設定読み込み (一部上書き) try: ws_tweet = self.sh.worksheet(self.SHEET_TWEET) settings.update(_parse_settings_from_sheet(ws_tweet)) except Exception as e: logger.warning(f"'{self.SHEET_TWEET}' の設定読み込み中にエラー: {e}") # 📨自ポスト管理シートからの設定読み込み (新機能用) try: ws_post = self.sh.worksheet(self.SHEET_POST) settings.update(_parse_settings_from_sheet(ws_post)) except Exception as e: logger.warning(f"'{self.SHEET_POST}' の設定読み込み中にエラー: {e}") logger.info(f"設定を取得しました: 設定項目数 {len(settings)}") return settings def _copy_row_validation(self, ws, source_row_idx: int, target_row_idx: int) -> None: """ 指定したシートの source_row_idx (1始まり) から target_row_idx (1始まり) へ 「データの入力規則（プルダウン、チェックボックス等）」のみをコピーする。 """ try: # gspread の sheetId (int) を取得 sheet_id = ws.id # Google Sheets API v4 の CopyPasteRequest を作成 # PASTE_DATA_VALIDATION を指定することで、値や書式を汚さず入力規則のみコピー可能 body = { "requests": [ { "copyPaste": { "source": { "sheetId": sheet_id, "startRowIndex": source_row_idx - 1, "endRowIndex": source_row_idx, "startColumnIndex": 0, "endColumnIndex": 20 # 十分な列数をカバー }, "destination": { "sheetId": sheet_id, "startRowIndex": target_row_idx - 1, "endRowIndex": target_row_idx, "startColumnIndex": 0, "endColumnIndex": 20 }, "pasteType": "PASTE_DATA_VALIDATION" } } ] } self.sh.batch_update(body) logger.debug(f"入力規則をコピーしました: {ws.title} {source_row_idx} -> {target_row_idx}") except Exception as e: logger.warning(f"入力規則のコピーに失敗しました (無視して続行): {e}") def _ensure_checkbox_validation(self, ws, start_row_idx: int, end_row_idx: int, col_idx: int) -> None: """ 指定範囲にチェックボックスの入力規則を設定する。 row_idx は1始まり、col_idx は0始まり。 """ if start_row_idx <= 0 or end_row_idx < start_row_idx: return try: body = { "requests": [ { "repeatCell": { "range": { "sheetId": ws.id, "startRowIndex": start_row_idx - 1, "endRowIndex": end_row_idx, "startColumnIndex": col_idx, "endColumnIndex": col_idx + 1, }, "cell": { "dataValidation": { "condition": {"type": "BOOLEAN"}, "strict": True, "showCustomUi": True, } }, "fields": "dataValidation", } } ] } self.sh.batch_update(body) logger.debug(f"チェックボックス入力規則を設定しました: {ws.title} R{start_row_idx}:R{end_row_idx} C{col_idx + 1}") except Exception as e: logger.warning(f"チェックボックス入力規則の設定に失敗しました (無視して続行): {e}") def get_registered_accounts(self) -> Dict[str, Dict]: """ アカウント管理シートのデータ開始行以降から登録済みアカウントリストを取得する """ try: ws = self.sh.worksheet(self.SHEET_ACCOUNT) data = ws.get_all_values() # データ開始行を動的検出（設定エリア終端 + ヘッダー1行分をスキップ） data_start = self._detect_data_start_row(ws) # 1始まり行番号 data_start_idx = data_start - 1 # 0始まりインデックス accounts = {} for i, row in enumerate(data[data_start_idx:], start=data_start): # COL_ACC_USERNAME の列が存在し、かつ空値でない場合 if len(row) > self.COL_ACC_USERNAME and row[self.COL_ACC_USERNAME].strip(): username = row[self.COL_ACC_USERNAME].strip() flg_1 = row[self.COL_ACC_FLG_REPLY].strip() if len(row) > self.COL_ACC_FLG_REPLY else "審査待ち" flg_final = row[self.COL_ACC_FLG_FOLLOW].strip() if len(row) > self.COL_ACC_FLG_FOLLOW else "" accounts[username] = { "row_index": i, "flg_1": flg_1, "flg_final": flg_final } return accounts except Exception as e: logger.error(f"登録済みアカウントの取得に失敗: {e}") return {} def _safe_str(self, val) -> str: """USER_ENTEREDで先頭の=等が数式として誤評価されないようエスケープする""" val_str = str(val) if val is not None else "" if val_str.startswith(("=", "+", "-", "@")): return "'" + val_str return val_str def upsert_account(self, account_data: Dict[str, str]) -> None: """ アカウント情報を書き込む。既に存在する場合はスキップ。 """ username = account_data.get("username") if not account_data.get("_skip_existing_check"): existing_accounts = self.get_registered_accounts() if username in existing_accounts: logger.info(f"Account {username} already exists; skipping.") return try: ws = self.sh.worksheet(self.SHEET_ACCOUNT) # 基準となるC列(アカウント名)のデータを取得し、最初の空白行を探す if self._account_insert_row_cache is None: c_col_data = ws.col_values(self.COL_ACC_USERNAME + 1) data_start = self._detect_data_start_row(ws) data_start_idx = data_start - 1 last_data_row = data_start - 1 for i in range(data_start_idx, len(c_col_data)): if str(c_col_data[i]).strip(): last_data_row = i + 1 insert_row_idx = max(data_start, last_data_row + 1) self._account_insert_row_cache = insert_row_idx self._account_data_start_cache = data_start else: insert_row_idx = self._account_insert_row_cache data_start = self._account_data_start_cache or self._detect_data_start_row(ws) self._account_data_start_cache = data_start logger.info(f"アカウント {username} を {insert_row_idx} 行目に追記します。") # A列から必要な最大列までをカバーする配列を生成（全てカラで初期化） max_col = max(self.COL_ACC_USERNAME, self.COL_ACC_PROFILE, self.COL_ACC_URL, self.COL_ACC_TIMESTAMP, self.COL_ACC_FLG_REPLY, self.COL_ACC_FLG_FOLLOW, self.COL_ACC_RESULT, self.COL_ACC_FOLLOW_RECOMMENDED, self.COL_ACC_ACCOUNT_SCORE, self.COL_ACC_SCORE_REASON, self.COL_ACC_EXCLUDE_REASON) row_data = [""] * (max_col + 1) # インデックスにしたがってデータを詰め込む row_data[self.COL_ACC_USERNAME] = self._safe_str(account_data.get("username", "")) # C列 row_data[self.COL_ACC_PROFILE] = self._safe_str(account_data.get("profile", "")) # D列 row_data[self.COL_ACC_URL] = account_data.get("profile_url", "") # E列 row_data[self.COL_ACC_TIMESTAMP] = account_data.get("timestamp", "") # F列 row_data[self.COL_ACC_FLG_REPLY] = account_data.get("flg_1") or "審査待ち" # G列 (デフォルト) # チェックボックス用 (文字列 "TRUE" ではなく Boolean にすることで確実にチェックされる) # 未指定(None) または 空文字("") の場合はデフォルトで True (チェックあり) flg_follow = account_data.get("flg_final") if flg_follow in [None, ""]: row_data[self.COL_ACC_FLG_FOLLOW] = True # デフォルト else: row_data[self.COL_ACC_FLG_FOLLOW] = True if str(flg_follow).upper() == "TRUE" else False row_data[self.COL_ACC_RESULT] = account_data.get("result", "待機中") # I列 row_data[self.COL_ACC_FOLLOW_RECOMMENDED] = account_data.get("follow_recommended", "") row_data[self.COL_ACC_ACCOUNT_SCORE] = account_data.get("account_score", "") row_data[self.COL_ACC_SCORE_REASON] = self._safe_str(account_data.get("score_reason", "")) row_data[self.COL_ACC_EXCLUDE_REASON] = self._safe_str(account_data.get("exclude_reason", "")) # A列から必要な列までを一括指定して更新 (例: A25:I25) # することで他の列にかぶらず、かつ絶対にズレない。 end_col_letter = chr(ord('A') + max_col) range_to_update = f"A{insert_row_idx}:{end_col_letter}{insert_row_idx}" # シートの最大行数を超えてしまう場合は、APIエラーを防ぐため行を事前に拡張する if insert_row_idx > ws.row_count: ws.add_rows(10) # 反映待ち import time time.sleep(1) # チェックボックスやプルダウンの入力規則を壊さないよう、USER_ENTERED を指定する # RAWだと文字列として上書きされフォーマットが剥がれる問題の対策 ws.update([row_data], range_to_update, value_input_option='USER_ENTERED') # 入力規則（プルダウン・チェックボックス）をデータ開始行（通常は設定のすぐ下）からコピーして修復 self._copy_row_validation(ws, data_start, insert_row_idx) self._account_insert_row_cache = insert_row_idx + 1 logger.info(f"アカウント {username} を書き込みました (列ズレなし)。") except Exception as e: logger.error(f"アカウントの書き込みに失敗しました: {e}", exc_info=True) def append_tweet(self, tweet_data: Dict[str, str], account_status: str) -> None: """ ツイート情報を書き込む。 """ if str(account_status).upper() == "FALSE": logger.info(f"アカウントステータスNGのため、ツイート {tweet_data.get('tweet_url')} の書き込みを除外します。") return try: ws = self.sh.worksheet(self.SHEET_TWEET) # 基準となるC列(アカウント名)のデータを取得 if self._tweet_insert_row_cache is None: c_col_data = ws.col_values(self.COL_TW_USERNAME + 1) data_start = self._detect_data_start_row(ws) data_start_idx = data_start - 1 last_data_row = data_start - 1 for i in range(data_start_idx, len(c_col_data)): if str(c_col_data[i]).strip(): last_data_row = i + 1 insert_row_idx = max(data_start, last_data_row + 1) self._tweet_insert_row_cache = insert_row_idx self._tweet_data_start_cache = data_start else: insert_row_idx = self._tweet_insert_row_cache data_start = self._tweet_data_start_cache or self._detect_data_start_row(ws) self._tweet_data_start_cache = data_start logger.info(f"ツイート {tweet_data.get('tweet_url')} を {insert_row_idx} 行目に追記します。") max_col = max(self.COL_TW_USERNAME, self.COL_TW_TEXT, self.COL_TW_URL, self.COL_TW_TIMESTAMP, self.COL_TW_ACC_STATUS, self.COL_TW_FLG_REPLY, self.COL_TW_AI_COMMENT, self.COL_TW_LIKE_RESULT, self.COL_TW_REPLY_RESULT, self.COL_TW_LIKE_RECOMMENDED, self.COL_TW_FOLLOW_RECOMMENDED, self.COL_TW_POST_SCORE, self.COL_TW_ACCOUNT_SCORE, self.COL_TW_SCORE_REASON, self.COL_TW_EXCLUDE_REASON) row_data = [""] * (max_col + 1) row_data[self.COL_TW_USERNAME] = self._safe_str(tweet_data.get("username", "")) row_data[self.COL_TW_TEXT] = self._safe_str(tweet_data.get("tweet_text", "")) row_data[self.COL_TW_URL] = tweet_data.get("tweet_url", "") row_data[self.COL_TW_TIMESTAMP] = tweet_data.get("timestamp", "") row_data[self.COL_TW_ACC_STATUS] = tweet_data.get("flg_1", "") # チェックボックス用 flg_reply = tweet_data.get("flg_2") if flg_reply is None: row_data[self.COL_TW_FLG_REPLY] = "" # デフォルト(手動チェック待ち) else: row_data[self.COL_TW_FLG_REPLY] = True if str(flg_reply).upper() == "TRUE" else False row_data[self.COL_TW_AI_COMMENT] = self._safe_str(tweet_data.get("ai_comment", "")) row_data[self.COL_TW_LIKE_RESULT] = tweet_data.get("like_result", "待機中") row_data[self.COL_TW_REPLY_RESULT] = tweet_data.get("reply_result", "") row_data[self.COL_TW_LIKE_RECOMMENDED] = tweet_data.get("like_recommended", "") # Follow recommendation is managed only on the account sheet; keep this legacy tweet column blank. row_data[self.COL_TW_FOLLOW_RECOMMENDED] = "" row_data[self.COL_TW_POST_SCORE] = tweet_data.get("post_score", "") row_data[self.COL_TW_ACCOUNT_SCORE] = tweet_data.get("account_score", "") row_data[self.COL_TW_SCORE_REASON] = self._safe_str(tweet_data.get("score_reason", "")) row_data[self.COL_TW_EXCLUDE_REASON] = self._safe_str(tweet_data.get("exclude_reason", "")) end_col_letter = chr(ord('A') + max_col) range_to_update = f"A{insert_row_idx}:{end_col_letter}{insert_row_idx}" # シートの最大行数を超えてしまう場合は、APIエラーを防ぐため行を事前に拡張する if insert_row_idx > ws.row_count: ws.add_rows(10) import time time.sleep(1) # エスケープ済みのためUSER_ENTEREDが安全に使用でき、チェックボックスの破壊を防ぐ ws.update([row_data], range_to_update, value_input_option='USER_ENTERED') # 入力規則をコピーして修復 self._copy_row_validation(ws, data_start, insert_row_idx) self._tweet_insert_row_cache = insert_row_idx + 1 logger.info(f"ツイート {tweet_data.get('tweet_url')} を追加しました。") except Exception as e: logger.error(f"ツイートの書き込みに失敗しました: {e}", exc_info=True) def get_reply_targets(self) -> List[Dict[str, Any]]: """ ツイート管理シートを走査し、「対象FLG②」が TRUE かつ 実行結果が未済の行を取得する。 """ targets = [] try: ws = self.sh.worksheet(self.SHEET_TWEET) data = ws.get_all_values() # データ開始行を動的検出 data_start = self._detect_data_start_row(ws) data_start_idx = data_start - 1 for i, row in enumerate(data[data_start_idx:], start=data_start): if len(row) > self.COL_TW_REPLY_RESULT: flg_reply = row[self.COL_TW_FLG_REPLY].strip() reply_result = row[self.COL_TW_REPLY_RESULT].strip() if flg_reply.upper() == "TRUE" and reply_result in ["", "待機中"]: targets.append({ "row_index": i, "username": row[self.COL_TW_USERNAME].strip(), "tweet_text": row[self.COL_TW_TEXT].strip(), "tweet_url": row[self.COL_TW_URL].strip(), "reply_text": row[self.COL_TW_AI_COMMENT].strip() if len(row) > self.COL_TW_AI_COMMENT else "" }) logger.info(f"リプライ実行対象を {len(targets)} 件見つけました。") return targets except Exception as e: logger.error(f"リプライ対象の取得に失敗しました: {e}") return targets def update_account_follow_flag(self, username: str, flag: bool) -> None: """ アカウント管理シートの指定したユーザー名のフォロー対象FLGを更新する。 """ existing_accounts = self.get_registered_accounts() if username not in existing_accounts: logger.warning(f"アカウント {username} が見つからないため、フォロー対象FLGの更新をスキップします。") return try: row_index = existing_accounts[username]["row_index"] ws = self.sh.worksheet(self.SHEET_ACCOUNT) # row_index は 0始まり なので +1 # COL_ACC_FLG_FOLLOW も 0始まり なので +1 して実際の列番号へ ws.update_cell(row_index + 1, self.COL_ACC_FLG_FOLLOW + 1, "TRUE" if flag else "FALSE") logger.info(f"アカウント {username} のフォロー対象FLGを {'TRUE' if flag else 'FALSE'} に更新しました。") except Exception as e: logger.error(f"アカウント {username} のフォロー対象FLGの更新に失敗しました: {e}") def update_tweet_reply_result(self, row_index: int, result_text: str) -> None: """ ツイート管理シートの特定行のリプライ実行結果を更新する。 """ try: ws = self.sh.worksheet(self.SHEET_TWEET) # 1始まりインデックスにするため +1 ws.update_cell(row_index, self.COL_TW_REPLY_RESULT + 1, result_text) logger.info(f"{row_index}行目のリプライ結果を '{result_text}' に更新しました。") except Exception as e: logger.error(f"行 {row_index} のリプライ結果の更新に失敗しました: {e}") def get_unprocessed_follow_targets(self, limit: int) -> List[Dict[str, Any]]: """ アカウント管理シートを走査し、「フォロー対象FLG」が TRUE かつ 実行結果が「未処理」の行を上から指定件数取得する。 """ targets = [] try: ws = self.sh.worksheet(self.SHEET_ACCOUNT) data = ws.get_all_values() data_start = self._detect_data_start_row(ws) data_start_idx = data_start - 1 for i, row in enumerate(data[data_start_idx:], start=data_start): if len(row) > self.COL_ACC_USERNAME: username = row[self.COL_ACC_USERNAME].strip() flg_follow = row[self.COL_ACC_FLG_FOLLOW].strip().upper() if len(row) > self.COL_ACC_FLG_FOLLOW else "" result = row[self.COL_ACC_RESULT].strip() if len(row) > self.COL_ACC_RESULT else "" if flg_follow == "TRUE" and result == "未処理": targets.append({ "row_index": i, "username": username }) if len(targets) >= limit: break logger.info(f"後追いフォロー実行対象を {len(targets)} 件見つけました。") return targets except Exception as e: logger.error(f"後追いフォロー対象の取得に失敗しました: {e}") return targets def update_account_result(self, username: str, result_text: str) -> None: """ アカウント管理シートの指定したユーザー名のフォロー実行結果を更新する。 """ existing_accounts = self.get_registered_accounts() if username not in existing_accounts: return try: row_index = existing_accounts[username]["row_index"] ws = self.sh.worksheet(self.SHEET_ACCOUNT) ws.update_cell(row_index + 1, self.COL_ACC_RESULT + 1, result_text) logger.info(f"アカウント {username} の結果を '{result_text}' に更新しました。") except Exception as e: logger.error(f"アカウント {username} の結果の更新に失敗しました: {e}") def get_post_targets(self, post_count: int = 1, random_mode: bool = False) -> List[Dict[str, Any]]: """ 自ポスト管理シートから投稿対象を取得する。 - random_mode=True: 有効FLGがTRUEの全行からランダムにpost_count件選択 - random_mode=False: 有効FLGがTRUE かつ 実行結果が空 の行を上からpost_count件選択 """ import random targets = [] try: ws = self.sh.worksheet(self.SHEET_POST) data = ws.get_all_values() data_start = self._detect_data_start_row(ws) data_start_idx = data_start - 1 candidates = [] for i, row in enumerate(data[data_start_idx:], start=data_start): if len(row) > self.COL_POST_ENABLE: text = row[self.COL_POST_TEXT].strip() enable = row[self.COL_POST_ENABLE].strip().upper() == "TRUE" result = row[self.COL_POST_RESULT].strip() if len(row) > self.COL_POST_RESULT else "" if not text or not enable: continue # 常に「未実行（resultが空）」の行のみを候補とする if not result: candidates.append({"row_index": i, "text": text}) # 候補をすべて見つける（シーケンシャルの場合も、ランダム抽選のために一度全リストを見る） if random_mode and candidates: # ランダムに抽出 targets = random.sample(candidates, min(len(candidates), post_count)) else: # シーケンシャルに抽出 targets = candidates[:post_count] logger.info(f"自ポスト投稿対象を {len(targets)} 件特定しました (モード: {'ランダム' if random_mode else 'シーケンシャル'})") return targets except Exception as e: logger.error(f"自ポスト対象の取得に失敗しました: {e}") return [] def update_post_result(self, row_index: int, result_text: str) -> None: """ 自ポスト管理シートの特定行に実行結果（日時等）を書き込む。 """ try: ws = self.sh.worksheet(self.SHEET_POST) ws.update_cell(row_index, self.COL_POST_RESULT + 1, result_text) logger.info(f"自ポスト{row_index}行目の実行結果を更新しました。") except Exception as e: logger.error(f"自ポスト{row_index}行目の更新に失敗しました: {e}") def get_post_generation_slots(self, limit: int) -> List[Dict[str, Any]]: """ 自ポスト管理シートから、AI生成本文を書き込む空き行を取得する。 C=キーワード, D=場面, E=補足 のいずれかが入力されている行、または G=有効FLG に TRUE/FALSE が入っている準備済み行を候補とする。 """ slots = [] if limit <= 0: return slots try: ws = self.sh.worksheet(self.SHEET_POST) data = ws.get_all_values() data_start = self._detect_data_start_row(ws) data_start_idx = data_start - 1 for i, row in enumerate(data[data_start_idx:], start=data_start): def _cell(col_idx: int) -> str: return row[col_idx].strip() if len(row) > col_idx else "" keywords = _cell(self.COL_POST_KEYWORDS) scene = _cell(self.COL_POST_SCENE) note = _cell(self.COL_POST_NOTE) text = _cell(self.COL_POST_TEXT) enable_raw = _cell(self.COL_POST_ENABLE).upper() result = _cell(self.COL_POST_RESULT) has_conditions = any([keywords, scene, note]) has_prepared_flag = enable_raw in ["TRUE", "FALSE"] if text or result: continue if not has_conditions and not has_prepared_flag: continue slots.append({ "row_index": i, "keywords": keywords, "scene": scene, "note": note, }) if len(slots) >= limit: break logger.info(f"AIポスト生成対象行を {len(slots)} 件取得しました。") return slots except Exception as e: logger.error(f"AIポスト生成対象行の取得に失敗しました: {e}", exc_info=True) return [] def write_post_drafts_to_slots(self, slots: List[Dict[str, Any]], post_texts: List[str]) -> None: """ 指定行の F列(本文), G列(有効FLG), H列(実行結果) にAI生成ポストを書き込む。 """ if not slots or not post_texts: logger.info("書き込むAIポストまたは対象行がありません。") return try: ws = self.sh.worksheet(self.SHEET_POST) count = min(len(slots), len(post_texts)) for slot, text in zip(slots[:count], post_texts[:count]): row_index = slot["row_index"] range_to_update = f"F{row_index}:H{row_index}" self._ensure_checkbox_validation(ws, row_index, row_index, self.COL_POST_ENABLE) ws.update( [[self._safe_str(text), True, ""]], range_to_update, value_input_option='USER_ENTERED' ) logger.info(f"AI生成ポストを {row_index} 行目へ書き込みました。") if len(post_texts) < len(slots): logger.warning(f"生成数が対象行数より少ないため、{len(slots) - len(post_texts)} 行は未更新です。") elif len(post_texts) > len(slots): logger.warning(f"生成数が対象行数より多いため、余剰 {len(post_texts) - len(slots)} 件は未使用です。") except Exception as e: logger.error(f"AI生成ポストの指定行書き込みに失敗しました: {e}", exc_info=True) def append_post_drafts(self, post_texts: List[str]) -> None: """ 生成されたポスト本文のリストを受け取り、自ポスト管理シートへ一行ずつ書き込む。 """ if not post_texts: logger.info("追加するポスト案がありません。") return try: ws = self.sh.worksheet(self.SHEET_POST) # 投稿文列のデータを取得し、最終行を探索 text_col_data = ws.col_values(self.COL_POST_TEXT + 1) data_start = self._detect_data_start_row(ws) data_start_idx = data_start - 1 insert_row_idx = data_start for i in range(data_start_idx, len(text_col_data)): if not str(text_col_data[i]).strip(): insert_row_idx = i + 1 break else: insert_row_idx = max(data_start, len(text_col_data) + 1) logger.info(f"AI生成のポスト {len(post_texts)} 件を {insert_row_idx} 行目から追記します。") max_col = max(self.COL_POST_TEXT, self.COL_POST_ENABLE, self.COL_POST_RESULT) rows_to_insert = [] for text in post_texts: row_data = [""] * (max_col + 1) row_data[self.COL_POST_TEXT] = self._safe_str(text) row_data[self.COL_POST_ENABLE] = True # 有効フラグをデフォルトでTRUE(チェック箱ON)に row_data[self.COL_POST_RESULT] = "" # 未実行 rows_to_insert.append(row_data) end_col_letter = chr(ord('A') + max_col) end_row_idx = insert_row_idx + len(rows_to_insert) - 1 range_to_update = f"A{insert_row_idx}:{end_col_letter}{end_row_idx}" # APIエラー回避のための行拡張 if end_row_idx > ws.row_count: ws.add_rows(len(rows_to_insert) + 10) import time time.sleep(1) # まとめて更新 ws.update(rows_to_insert, range_to_update, value_input_option='USER_ENTERED') # 入力規則（チェックボックスなど）を各行にコピー for i in range(len(rows_to_insert)): self._copy_row_validation(ws, data_start, insert_row_idx + i) self._ensure_checkbox_validation(ws, insert_row_idx, end_row_idx, self.COL_POST_ENABLE) logger.info(f"{len(post_texts)} 件のAIポストをスプレッドシートに書き込み完了しました。") except Exception as e: logger.error(f"自ポストの書き込み(append)に失敗しました: {e}", exc_info=True) if __name__ == "__main__": # テスト用 logging.basicConfig(level=logging.INFO) 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() accs = manager.get_registered_accounts() print(f"\n--- 登録済みアカウント: {len(accs)}件 ---") print(list(accs.keys()))