import os import sys import time import json import urllib.request import urllib.error import logging vendor_path = os.path.join(os.path.dirname(__file__), 'vendor') if vendor_path not in sys.path: sys.path.insert(0, vendor_path) from x_sheets_writer import XSheetsManager # ログの設定 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger("x_post_generator") CREDS_FILE = os.path.join(os.path.dirname(__file__), "credentials.json") SPREADSHEET_ID = "1icqi5ZjqjRUOcGnxtDyHw1-gXj-osmGjB8twUnKB6i0" def build_condition_block(slots: list, total_count: int = 0) -> str: """シート上の作成条件を、AIに渡しやすいテキストへ整形する。""" if not slots: return "" lines = [ "【今回の行別作成条件】", "以下の各行について、1行につき1投稿を作成してください。", "キーワード・場面・補足は空欄の場合があります。空欄の項目は無理に補わず、共通プロンプトに従って自然に作成してください。", "キーワードは必ず全語を本文に入れるという意味ではなく、題材・文脈として反映してください。", "場面が入力されている場合は、その場面が伝わる具体的な一瞬から書いてください。", "補足が入力されている場合は、他の条件より優先して反映してください。", "", ] for idx, slot in enumerate(slots, start=1): keywords = slot.get("keywords") or "（指定なし）" scene = slot.get("scene") or "（指定なし）" note = slot.get("note") or "（指定なし）" row_index = slot.get("row_index", "") row_label = f" / シート{row_index}行目" if row_index else "" lines.extend([ f"{idx}.{row_label}", f"キーワード: {keywords}", f"場面: {scene}", f"補足: {note}", "", ]) if total_count and len(slots) < total_count: remaining = total_count - len(slots) lines.extend([ f"上記の行別作成条件に対応する投稿を {len(slots)} 件作成したあと、残り {remaining} 件は共通プロンプトに従って作成してください。", "残りの投稿は、行別条件の投稿とテーマ、場面、相手、困り方、始まり方、終わり方が重ならないようにしてください。", "", ]) lines.extend([ "【今回分の多様性ルール】", "・今回作成する投稿同士で、テーマ、場面、相手、困り方、始まり方、終わり方をできるだけ重ねないでください。", "・条件が空欄の行は、他の行と似ない題材を選んでください。", "・「いい感じにしといて」「急な予定変更」「大丈夫？」など、直近で多くなりやすい型に安易に寄せないでください。", ]) return "\n".join(lines) def parse_generated_posts(result_text: str, post_count: int) -> list: """Geminiの返答から投稿本文リストを取り出す。JSON配列を優先する。""" import re text = (result_text or "").strip() if not text: return [] if text.startswith("```"): text = re.sub(r"^```(?:json)?\s*", "", text) text = re.sub(r"\s*```$", "", text).strip() json_candidates = [text] match = re.search(r"\[[\s\S]*\]", text) if match: json_candidates.append(match.group(0)) for candidate in json_candidates: try: parsed = json.loads(candidate) if isinstance(parsed, list): return [str(item).strip().replace("\n", " ") for item in parsed if str(item).strip()] if isinstance(parsed, dict): for key in ["posts", "items", "results"]: value = parsed.get(key) if isinstance(value, list): return [str(item).strip().replace("\n", " ") for item in value if str(item).strip()] except Exception: pass posts = [] for line in text.splitlines(): line = line.strip() if len(line) < 5 or line.startswith("```") or line.startswith("---"): continue line = re.sub(r"^\d+[\.\:：]\s*", "", line) line = re.sub(r"^投稿案\d*[\.\:：]\s*", "", line) if line: posts.append(line) return posts[:post_count] def generate_posts(api_key: str, post_count: int, base_prompt: str, slots: list = None) -> list: """Gemini API を使ってポスト文を自動生成する（追加ライブラリに依存しないREST版）""" condition_block = build_condition_block(slots or [], post_count) # スプレッドシートで設定されたプロンプトに、プログラムが処理するための「文字の出力形式」と「作成数」の厳格なルールを付与します。 prompt = f""" {base_prompt} {condition_block} ------------------- 【システムからの必須出力ルール】 ・必ず【計 {post_count} 投稿分】を作成すること。 ・出力はJSON配列のみとすること。 ・JSON配列の要素数は必ず {post_count} 個にすること。 ・各要素は1投稿分の本文文字列にすること。 ・投稿本文の途中に改行を入れないこと。 ・行頭番号、「投稿案：」、説明文、Markdown、コードブロックは出力しないこと。 """ try: logger.info(f"Gemini API に {post_count} 件のポスト生成をリクエストします...") # google-genai ライブラリを使わず、標準の通信機能だけでGeminiを呼び出す url = f"https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-flash:generateContent?key={api_key}" headers = {'Content-Type': 'application/json'} data = { "contents": [{"parts": [{"text": prompt}]}], "generationConfig": { "temperature": 0.85, "responseMimeType": "application/json" } } req = urllib.request.Request(url, data=json.dumps(data).encode('utf-8'), headers=headers, method='POST') with urllib.request.urlopen(req) as response: result_data = json.loads(response.read().decode('utf-8')) result_text = result_data['candidates'][0]['content']['parts'][0]['text'] result_text = result_text.strip() if result_text else "" posts = parse_generated_posts(result_text, post_count) logger.info(f"{len(posts)} 件のポストを生成しました。") return posts except urllib.error.URLError as e: logger.error(f"Gemini API 通信エラー: {e}") if hasattr(e, 'read'): logger.error(e.read().decode('utf-8')) return [] except Exception as e: logger.error(f"AIコメントの生成中にエラーが発生しました: {e}", exc_info=True) return [] def main(): logger.info("ポスト自動生成プログラムを開始します。") try: manager = XSheetsManager(CREDS_FILE, SPREADSHEET_ID) settings = manager.get_settings() api_key = settings.get("gemini_api_key", "").strip() if not api_key: api_key = os.environ.get("GEMINI_API_KEY", "") if not api_key: logger.error("エラー: スプレッドシート (gemini_api_key) にAPIキーが設定されていません。") return post_count_str = settings.get("self_post_times", 6) try: post_count = int(post_count_str) except ValueError: post_count = 6 logger.warning(f"self_post_times の値 '{post_count_str}' が数値でないため、デフォルトの作成数 {post_count} で実行します。") if post_count <= 0: logger.warning("作成数(self_post_times)が0以下のため、生成をスキップします。") return base_prompt = settings.get("self_post_prompt", "").strip() if not base_prompt: logger.error("エラー: スプレッドシートの設定欄(self_post_prompt)にプロンプトが入力されていません。") return requested_post_count = post_count generation_slots = manager.get_post_generation_slots(requested_post_count) if generation_slots: remaining_count = requested_post_count - len(generation_slots) if remaining_count > 0: logger.info(f"シート上の作成条件 {len(generation_slots)} 行と、通常生成 {remaining_count} 件でポストを生成します。") else: logger.info(f"シート上の作成条件 {len(generation_slots)} 行を使ってポストを生成します。") else: logger.info("作成条件の空き行が見つからないため、従来どおり末尾へ追記します。") generated_posts = generate_posts(api_key, requested_post_count, base_prompt, generation_slots) if generated_posts: if generation_slots: slot_count = len(generation_slots) manager.write_post_drafts_to_slots(generation_slots, generated_posts[:slot_count]) remaining_posts = generated_posts[slot_count:] if remaining_posts: manager.append_post_drafts(remaining_posts) else: manager.append_post_drafts(generated_posts) logger.info("プログラムが正常に終了しました。") else: logger.warning("対象となるポストが生成されなかったため、スプレッドシートへの追記は行いませんでした。") except Exception as e: logger.error(f"プログラム実行中に予期せぬエラーが発生しました: {e}", exc_info=True) if __name__ == "__main__": main()