# [メタ情報] # 識別子: XXXXXXブログgem作成パイプライン_exe # 補足: env終了 # [/メタ情報] 要約: このテキスト群は、XXXXXX.comのブログ記事を自動的に取得、加工、そして複数のプラットフォームへ同期させる一連のシステムを記述しています。 メインの`update_all.sh`スクリプトは、WordPress APIから記事のID、タイトル、更新日、本文などを取得(`XXXXXX_full_auto.py`)し、不要なHTMLタグを除去した「ラベル付きブロック形式」の統合TXTデータ(`XXXXXX_clean_database.txt`)を作成(`docx_to_clean_csv.py`)後、Googleドライブへアップロードします。 さらに、`split_db.py`は、この統合TXTを更新日とIDでソートし、指定文字数ごとに最大10個のローカル分割ファイルとして出力します。また、`sync_blog_docs.py`は同様の処理を行いながら、ソートされた記事データを直接10個の指定されたGoogleドキュメントに上書き同期します。 これらの自動化処理は、macOSのLaunchAgent設定ファイル(`com.XXXXXX.updateall.plist`と`com.XXXXXX.syncblog.plist`)により、それぞれ毎日午前4時と午前5時に自動実行されるようにスケジュールされており、ブログ記事データの継続的な収集、整理、複数サービスへの反映を目的としています。 /Users/XXXXXX/python_scripts/update_all.sh update_all.sh ``` #!/bin/bash # ========================================== # 【設定】ホームディレクトリの .env からパスを動的ロード # ========================================== ENV_FILE="$HOME/python_scripts/.env" if [ -f "$ENV_FILE" ]; then # tr -d '\r' でWindowsの改行コード(CR)を除去してループ処理 while read -r line || [ -n "$line" ]; do # 1. 『=』が含まれていない行は、メモや不正な行として「最初から無視」 if [[ "$line" != *"="* ]]; then continue fi # 2. 行頭が「#」で始まる純粋なコメント行も無視 if [[ "$line" =~ ^[[:space:]]*# ]]; then continue fi # 3. 行の途中に「#」がある場合、コメントとしてそれ以降をカット clean_line="${line%%#*}" # 4. bashの機能で前後空白をトリミング trimmed="${clean_line#"${clean_line%%[![:space:]]*}"}" trimmed="${trimmed%"${trimmed##*[![:space:]]}"}" # 5. 変数名として正しい「KEY=VALUE」のパターンのみ処理 if [[ "$trimmed" =~ ^[A-Za-z_][A-Za-z0-9_]*= ]]; then # 6. クォーテーション(" や ')をロード段階で完全に一括除去 clean_env_line=$(echo "$trimmed" | tr -d '"' | tr -d "'") # 7. KEY と VALUE に安全に分割 key="${clean_env_line%%=*}" val="${clean_env_line#*=}" # 8. 直接エクスポート(値にスペースがあっても安全に代入) export "$key"="$val" fi done < <(tr -d '\r' < "$ENV_FILE") else echo "❌ エラー: .env ファイルが見つかりません。 ($ENV_FILE)" >&2 exit 1 fi # ========================================== # 環境変数の揺れ(単数形・複数形・接頭辞の有無)を網羅的に吸収 # ========================================== SCRIPTS_DIR_CANDIDATE="${XXXXXX_PYTHON_SCRIPTS_DIR:-${XXXXXX_PYTHON_SCRIPT_DIR:-${PYTHON_SCRIPTS_DIR:-${PYTHON_SCRIPT_DIR}}}}" # 【究極の自律解決フォールバック】 # .envに何も定義されていなければ、このシェルスクリプトが存在する親ディレクトリ(/Users/XXXXXX/python_scripts)を自動的に採用します SELF_DIR="$(cd "$(dirname "$0")" && pwd)" SCRIPTS_DIR="${SCRIPTS_DIR_CANDIDATE:-$SELF_DIR}" # 起動前検証(自律停止ガード) if [ -z "$XXXXXX_BLOG_DRIVE_PATH" ] || [ -z "$XXXXXX_BLOG_DB_FILE" ] || [ -z "$SCRIPTS_DIR" ]; then echo "❌ エラー:必要な環境変数が .env に設定されていません。" >&2 echo "ロードされた環境変数リスト:" echo " - XXXXXX_BLOG_DRIVE_PATH: ${XXXXXX_BLOG_DRIVE_PATH:-(未設定)}" echo " - XXXXXX_BLOG_DB_FILE: ${XXXXXX_BLOG_DB_FILE:-(未設定)}" echo " - 実行ディレクトリ(SCRIPTS_DIR): ${SCRIPTS_DIR:-(未設定)}" exit 1 fi DRIVE_PATH="$XXXXXX_BLOG_DRIVE_PATH" DB_FILE="$XXXXXX_BLOG_DB_FILE" # ========================================== # 処理開始 (マスターパイプライン) # ========================================== cd "$SCRIPTS_DIR" || { echo "❌ エラー: スクリプトディレクトリへの移動に失敗しました。 ($SCRIPTS_DIR)"; exit 1; } echo "=== 【第1工程】ブログ記事の取得 (XXXXXX_full_auto.py) ===" /usr/bin/python3 XXXXXX_full_auto.py echo "" echo "=== 【第2工程】統合TXTデータの作成 (docx_to_clean_csv.py) ===" /usr/bin/python3 docx_to_clean_csv.py echo "" echo "=== 【第3工程】統合TXTデータベースの10分割スライス (split_db.py) ===" /usr/bin/python3 split_db.py echo "" echo "=== 【第4工程】Googleドライブへの統合TXTのバックアップコピー ===" # フォルダがあるか確認してからコピー if [ -d "$DRIVE_PATH" ]; then cp "$DB_FILE" "$DRIVE_PATH/" echo "✅ 成功!Googleドライブに最新の統合TXTを配置しました。" echo "保存先: $DRIVE_PATH" else echo "❌ エラー: 指定されたGoogleドライブの場所が見つかりません。" echo "設定したパス: $DRIVE_PATH" exit 1 fi echo "" echo "=== 【第5工程】Googleドキュメント(10枠)への上書き同期 (sync_blog_docs.py) ===" /usr/bin/python3 sync_blog_docs.py echo "" echo "🎉 全5工程のマスターパイプライン処理が完全完了しました! 🎉" ``` /Users/XXXXXX/python_scripts/XXXXXX_full_auto.py XXXXXX_full_auto.py ``` import os import sys import requests import re import time from pathlib import Path from dotenv import load_dotenv # ホームディレクトリの .env を確実にロード ENV_PATH = Path.home() / "python_scripts" / ".env" load_dotenv(dotenv_path=ENV_PATH) # 環境変数の存在検証(起動時自律停止ガード) REQUIRED_ENV_VARS = ["XXXXXX_BLOG_SITE_URL", "XXXXXX_BLOG_ARTICLES_DIR"] missing_vars = [var for var in REQUIRED_ENV_VARS if not os.getenv(var)] if missing_vars: print(f"エラー:必要な鍵が.envに設定されていません: {', '.join(missing_vars)}", file=sys.stderr) sys.exit(1) SITE_URL = os.getenv("XXXXXX_BLOG_SITE_URL") SAVE_DIR = os.getenv("XXXXXX_BLOG_ARTICLES_DIR") def sync_articles_with_true_date(): api_url = f"{SITE_URL}/wp-json/wp/v2/posts" page = 1 total_synced = 0 if not os.path.exists(SAVE_DIR): os.makedirs(SAVE_DIR) print("🚀 サーバーから『真の更新日』を取得中...") while True: try: # サーバー負荷を抑えるため20件ずつ params = {'per_page': 20, 'page': page} response = requests.get(api_url, params=params, timeout=60) if response.status_code != 200: if response.status_code == 400: # ページがなくなったら終了 break print(f"\n⚠️ サーバーエラー: {response.status_code}") break except Exception as e: print(f"\n⚠️ 接続エラー: {e}") break posts = response.json() if not posts: break for post in posts: post_id = post['id'] # modified の先頭10文字 (2026-05-12) を取って、ハイフンを点に変える true_update_date = post['modified'][:10].replace('-', '.') pub_date = post['date'][:10].replace('-', '.') title = post['title']['rendered'] safe_title = re.sub(r'[\\/:*?"<>|]', '_', title) # ファイル名は [ID-番号] 形式 filename = f"[ID-{post_id}] {safe_title}.txt" filepath = os.path.join(SAVE_DIR, filename) lines = [ f"タイトル: {title}", f"[ID-{post_id}]", f"更新日: {true_update_date}", f"公開日: {pub_date}", f"URL: {post['link']}", "-" * 30, post['content']['rendered'] ] with open(filepath, 'w', encoding='utf-8') as f: f.write("\n".join(lines)) total_synced += 1 print(f"✅ {page}ページ目(累計 {total_synced}件)を取得完了...", end='\r') page += 1 time.sleep(0.5) # サーバーをいたわる休憩 if total_synced > 0: print(f"\n\n✨ 成功!合計 {total_synced} 件 of ファイルを『正確な更新日』で同期しました。") else: print("\n\n🤔 取得できませんでした。") if __name__ == "__main__": sync_articles_with_true_date() ``` /Users/XXXXXX/python_scripts/check_dates.py ``` import requests from bs4 import BeautifulSoup import re import time # URLの末尾に「?t=現在時刻」をつけて、サーバーキャッシュを100%貫通します target_url = "https://XXXXXX.com/2026/04/18/28865/" buster_url = f"{target_url}?t={time.time()}" print(f"🧨 サーバーキャッシュ強制貫通テストを開始します...\n") print(f"アクセス先: {buster_url}\n") try: res = requests.get(buster_url, headers={'User-Agent': 'Mozilla/5.0'}, timeout=10) res.encoding = 'utf-8' soup = BeautifulSoup(res.text, 'html.parser') print(" [メタデータ(検索エンジン用)]") metas = soup.find_all('meta', property=re.compile(r'(published_time|modified_time)')) for m in metas: print(f" - {m.get('property')}: {m.get('content')}") except Exception as e: print(f" エラーが発生しました: {e}") ``` /Users/XXXXXX/python_scripts/docx_to_clean_csv.py docx_to_clean_csv.py ``` import os import sys import re from pathlib import Path from dotenv import load_dotenv # ホームディレクトリの .env を確実にロード ENV_PATH = Path.home() / "python_scripts" / ".env" load_dotenv(dotenv_path=ENV_PATH) # 環境変数の存在検証(起動時自律停止ガード) REQUIRED_ENV_VARS = ["XXXXXX_BLOG_ARTICLES_DIR", "XXXXXX_BLOG_DB_FILE"] missing_vars = [var for var in REQUIRED_ENV_VARS if not os.getenv(var)] if missing_vars: print(f"エラー:必要な鍵が.envに設定されていません: {', '.join(missing_vars)}", file=sys.stderr) sys.exit(1) SOURCE_DIR = os.getenv("XXXXXX_BLOG_ARTICLES_DIR") OUTPUT_FILE = os.getenv("XXXXXX_BLOG_DB_FILE") def main(): if not os.path.exists(SOURCE_DIR): print(f"❌ フォルダが見つかりません: {SOURCE_DIR}") return files = sorted([f for f in os.listdir(SOURCE_DIR) if f.endswith('.txt')]) print(f"🚀 {len(files)}件を「ラベル付きブロック形式」で統合中...") with open(OUTPUT_FILE, 'w', encoding='utf-8') as f_out: for filename in files: try: with open(os.path.join(SOURCE_DIR, filename), 'r', encoding='utf-8') as f_in: content = f_in.read() # --- 項目抽出 --- # ID抽出 id_match = re.search(r'\[ID-(.*?)\]', filename) id_val = id_match.group(1) if id_match else 'N/A' # タイトル抽出 # 正規表現 \[ID-.*?\] を使ってファイル名からIDタグを消す clean_title = re.sub(r'\[ID-.*?\]', '', filename).replace('.txt', '').strip() # 更新日・公開日 u_match = re.search(r'更新日:\s*([\d\.]+)', content) u_date = u_match.group(1) if u_match else 'N/A' p_match = re.search(r'公開日:\s*([\d\.]+)', content) p_date = p_match.group(1) if p_match else 'N/A' # URL url_match = re.search(r'URL:\s*(https?://[^\s\n]+)', content) url_val = url_match.group(1) if url_match else 'N/A' # 本文抽出(線より後) if "------------------------------" in content: # 分割を1回に制限し、最初の区切り線より後ろをすべて取得する parts = content.split("------------------------------", 1) pure_body = parts[1].strip() if len(parts) > 1 else content else: pure_body = content # --- 不要なタグのクリーニング処理 --- # を中身ごと削除(大文字小文字区別なし、改行をまたぐ) pure_body = re.sub(r'.*?', '', pure_body, flags=re.DOTALL | re.IGNORECASE) # を中身ごと削除 pure_body = re.sub(r'.*?', '', pure_body, flags=re.DOTALL | re.IGNORECASE) # --- 書き出し --- f_out.write("[[[ARTICLE_START]]]\n") f_out.write(f"ID:[ID-{id_val}]\n") f_out.write(f"TITLE:{clean_title}\n") f_out.write(f"UPDATE:{u_date}\n") f_out.write(f"POST:{p_date}\n") f_out.write(f"URL:{url_val}\n") f_out.write("BODY_START:\n") f_out.write(pure_body + "\n") f_out.write("[[[ARTICLE_END]]]\n\n") except Exception as e: print(f"⚠️ スキップ: {filename} ({e})") print(f"✅ 統合完了: {OUTPUT_FILE}") if __name__ == "__main__": main() ``` /Users/XXXXXX/python_scripts/split_db.py ``` import re import os import sys from pathlib import Path from dotenv import load_dotenv # ホームディレクトリの .env を確実にロード ENV_PATH = Path.home() / "python_scripts" / ".env" load_dotenv(dotenv_path=ENV_PATH) # 環境変数の存在検証(起動時自律停止ガード) REQUIRED_ENV_VARS = ["XXXXXX_BLOG_DB_FILE", "XXXXXX_BLOG_PART_PREFIX"] missing_vars = [var for var in REQUIRED_ENV_VARS if not os.getenv(var)] if missing_vars: print(f"エラー:必要な鍵が.envに設定されていません: {', '.join(missing_vars)}", file=sys.stderr) sys.exit(1) input_file = os.getenv("XXXXXX_BLOG_DB_FILE") output_prefix = os.getenv("XXXXXX_BLOG_PART_PREFIX") chars_per_file = 350000 # 1ファイルあたりの目安文字数(約35万文字) MAX_FILES = 10 def process_database(): print("データの読み込みと解析を開始します...") # 1. ファイルの読み込み try: with open(input_file, 'r', encoding='utf-8') as f: content = f.read() except FileNotFoundError: print(f"エラー: '{input_file}' が見つかりません。パスが正しいか確認してください。", file=sys.stderr) return # 2. 記事ブロックの抽出 pattern = r'(\[\[\[ARTICLE_START\]\]\].*?\[\[\[ARTICLE_END\]\]\])' blocks = re.findall(pattern, content, re.DOTALL) articles = [] for block in blocks: update_match = re.search(r'UPDATE:\s*([0-9\.]+)', block) id_match = re.search(r'ID:\[.*?([0-9]+)\]', block) update_val = update_match.group(1) if update_match else "0000.00.00" id_val = int(id_match.group(1)) if id_match else 0 articles.append({ 'update': update_val, 'id': id_val, 'content': block }) # 3. ソート(更新日降順 > ID降順) articles.sort(key=lambda x: (x['update'], x['id']), reverse=True) # 4. 目次番号の付与とファイル分割 current_file_index = 1 current_char_count = 0 current_file_content = [] for i, article in enumerate(articles): index_no = i + 1 index_str = f"INDEX: No.{index_no:03}\n" modified_content = re.sub(r'(ID:\[.*?\])', rf'{index_str}\1', article['content'], count=1) current_file_content.append(modified_content) current_char_count += len(modified_content) # 5. 文字数が目安を超え、かつMAX_FILES未満なら新しいファイルへ if current_char_count >= chars_per_file and current_file_index < MAX_FILES: write_file(current_file_index, current_file_content) current_file_index += 1 current_char_count = 0 current_file_content = [] # 6. 余った最後のブロックを書き出し if current_file_content: write_file(current_file_index, current_file_content) current_file_index += 1 # 7. MAX_FILES(10個)に満たない場合は、空のファイルを生成して埋める while current_file_index <= MAX_FILES: write_file(current_file_index, ["(このドキュメントに割り当てられるファイルはありません)"]) current_file_index += 1 print(f"\n🎉 完了しました!仕様通り、全 {MAX_FILES} 個のファイルを出力しました。") def write_file(index, content_list): filename = f"{output_prefix}{index}.txt" with open(filename, 'w', encoding='utf-8') as f: f.write("\n\n".join(content_list)) char_len = sum(len(c) for c in content_list) if content_list == ["(このドキュメントに割り当てられるファイルはありません)"]: print(f" -> {filename} を作成しました(※空ファイルとして処理)") else: print(f" -> {filename} を作成しました(約 {char_len:,} 文字)") if __name__ == "__main__": process_database() ``` /Users/XXXXXX/python_scripts/sync_blog_docs.py ``` import os import re import sys import socket from pathlib import Path from dotenv import load_dotenv from googleapiclient.discovery import build from google.auth.transport.requests import Request from google.oauth2.credentials import Credentials from google_auth_oauthlib.flow import InstalledAppFlow socket.setdefaulttimeout(600) # ホームディレクトリの .env を確実にロード ENV_PATH = Path.home() / "python_scripts" / ".env" load_dotenv(dotenv_path=ENV_PATH) # 動的な TARGET_DOC_IDS のバインドと検証 TARGET_DOC_IDS = [] missing_ids = [] for idx in range(1, 11): env_key = f"XXXXXX_BLOG_DOC_ID_{idx}" doc_id = os.getenv(env_key) if not doc_id: missing_ids.append(env_key) else: TARGET_DOC_IDS.append(doc_id) REQUIRED_ENV_VARS = ["XXXXXX_BLOG_DB_FILE", "XXXXXX_GOOGLE_CREDENTIALS_FILE", "XXXXXX_GOOGLE_TOKEN_FILE"] missing_vars = [var for var in REQUIRED_ENV_VARS if not os.getenv(var)] # 設定漏れがある場合は自律停止ガードをトリガー if missing_ids or missing_vars: all_missing = missing_ids + missing_vars print(f"エラー:必要な鍵が.envに設定されていません: {', '.join(all_missing)}", file=sys.stderr) sys.exit(1) INPUT_FILE = os.getenv("XXXXXX_BLOG_DB_FILE") CLIENT_SECRET_FILE = os.getenv("XXXXXX_GOOGLE_CREDENTIALS_FILE") TOKEN_FILE = os.getenv("XXXXXX_GOOGLE_TOKEN_FILE") SCOPES = ['https://www.googleapis.com/auth/documents'] CHARS_PER_FILE = 350000 # 1ドキュメントあたりの上限目安文字数 def get_credentials(): creds = None if os.path.exists(TOKEN_FILE): creds = Credentials.from_authorized_user_file(TOKEN_FILE, SCOPES) if not creds or not creds.valid: if creds and creds.expired and creds.refresh_token: print("🔄 認証トークンを更新中...") creds.refresh(Request()) else: print("🌍 ブラウザを開いてログインしてください...") flow = InstalledAppFlow.from_client_secrets_file(CLIENT_SECRET_FILE, SCOPES) creds = flow.run_local_server(port=0) with open(TOKEN_FILE, 'w') as token: token.write(creds.to_json()) return creds def update_google_doc(service, target_doc_id, text_content): short_id = target_doc_id[-5:] print(f" -> Google Docs APIに接続中... (ID: ...{short_id})") try: doc = service.documents().get(documentId=target_doc_id, fields="body(content(endIndex))").execute() content = doc.get('body').get('content') end_index = content[-1]['endIndex'] - 1 if content else 1 requests = [] if end_index > 1: requests.append({'deleteContentRange': {'range': {'startIndex': 1, 'endIndex': end_index}}}) requests.append({'insertText': {'location': {'index': 1}, 'text': text_content}}) service.documents().batchUpdate(documentId=target_doc_id, body={'requests': requests}).execute() print(f" ✅ 更新完了 (ID: ...{short_id})") except Exception as e: print(f" ❌ APIエラー (ID: ...{short_id}): {e}") def process_and_sync(): print("データの読み込みと解析を開始します...") try: with open(INPUT_FILE, 'r', encoding='utf-8') as f: content = f.read() except FileNotFoundError: print(f"エラー: '{INPUT_FILE}' が見つかりません。", file=sys.stderr) return # 記事ブロックの抽出 pattern = r'(\[\[\[ARTICLE_START\]\]\].*?\[\[\[ARTICLE_END\]\]\])' blocks = re.findall(pattern, content, re.DOTALL) articles = [] for block in blocks: update_match = re.search(r'UPDATE:\s*([0-9\.]+)', block) id_match = re.search(r'ID:\[.*?([0-9]+)\]', block) articles.append({ 'update': update_match.group(1) if update_match else "0000.00.00", 'id': int(id_match.group(1)) if id_match else 0, 'content': block }) # ソート(UPDATE降順 > ID降順) articles.sort(key=lambda x: (x['update'], x['id']), reverse=True) # API接続準備 creds = get_credentials() service = build('docs', 'v1', credentials=creds) # 分割と書き込み current_doc_index = 0 current_buffer = [] current_char_count = 0 overall_count = 1 print("\nGoogleドキュメントへの同期を開始します...") for art in articles: index_str = f"INDEX: No.{overall_count:03}\n" modified_content = re.sub(r'(ID:\[.*?\])', rf'{index_str}\1', art['content'], count=1) overall_count += 1 if (current_char_count + len(modified_content) > CHARS_PER_FILE) and (current_doc_index < 9): text_to_write = "\n\n".join(current_buffer) print(f"\n--- Doc #{current_doc_index + 1} の更新 ---") update_google_doc(service, TARGET_DOC_IDS[current_doc_index], text_to_write) current_doc_index += 1 current_buffer = [] current_char_count = 0 current_buffer.append(modified_content) current_char_count += len(modified_content) # 残りのデータ書き込み print(f"\n--- Doc #{current_doc_index + 1} の更新 ---") update_google_doc(service, TARGET_DOC_IDS[current_doc_index], "\n\n".join(current_buffer) or "(このドキュメントに割り当てられるファイルはありません)") current_doc_index += 1 # 余った枠を空ファイルとして処理 for i in range(current_doc_index, 10): print(f"\n--- Doc #{i + 1} の更新 (空ファイル) ---") update_google_doc(service, TARGET_DOC_IDS[i], "(このドキュメントに割り当てられるファイルはありません)") print("\n🎉 すべての同期が完了しました!") if __name__ == "__main__": process_and_sync() ``` /Users/XXXXXX/Library/LaunchAgents/com.XXXXXX.updateall.plist ``` Label com.XXXXXX.updateall ProgramArguments /bin/bash /Users/XXXXXX/python_scripts/update_all.sh StartCalendarInterval Hour 4 Minute 0 StandardOutPath /Users/XXXXXX/Library/Logs/update_all.log StandardErrorPath /Users/XXXXXX/Library/Logs/update_all.err ``` /Users/XXXXXX/Library/LaunchAgents/com.XXXXXX.syncblog.plist は以下だが、パイプライン化したので停止。一応、フォルダに残しておく。 ``` Label com.XXXXXX.syncblog ProgramArguments /usr/bin/python3 /Users/XXXXXX/python_scripts/sync_blog_docs.py StartCalendarInterval Hour 5 Minute 0 StandardOutPath /Users/XXXXXX/Library/Logs/sync_blog_docs.log StandardErrorPath /Users/XXXXXX/Library/Logs/sync_blog_docs.err ``` XXXXXXブログ今すぐ更新_M1Mac専用.app AppleScript ``` -- 一本化されたマスターパイプライン(update_all.sh)を1回だけ呼び出して安全に全自動完走させます。 tell application "Terminal" activate do script "/bin/bash /Users/XXXXXX/python_scripts/update_all.sh" end tell ```