X美しい未来は美しい心から投稿版（最終）

/home/users/xxxx/xxxx/web/beautiful_future/ config.py post_ja_x_ig.py # 日本語X + Instagram 同時投稿 post_en_x.py # 英語X投稿 post_zh_x.py # 中国語X投稿 data/ ja_posts.csv # 日本語＋祝日版CSV multi_lang_posts.csv # 英語・中国語版CSV logs/ post.log # 共通ログ public_images/ # Web公開ディレクトリ配下に配置推奨 morning/ ig_morning_01_yellow_sunrise.png ... ig_morning_10_yellow_smile.png noon/ ig_noon_01_blue_sky.png ... ig_noon_10_yellow_focus.png night/ ig_night_01_navy_calm.png ... ig_night_10_purple_moonlight.png

public_images/ morning/ ig_morning_01_yellow_sunrise.png ... ig_morning_10_yellow_smile.png noon/ ig_noon_01_blue_sky.png ... ig_noon_10_yellow_focus.png night/ ig_night_01_navy_calm.png ... ig_night_10_purple_moonlight.png

ig_{slot}_{seq}_{color}_{keyword}.png

# config.py # 共通設定ファイル import os from datetime import timedelta, timezone # ===== タイムゾーン ===== JST = timezone(timedelta(hours=9)) # ===== ベースディレクトリ ===== BASE_DIR = os.path.dirname(os.path.abspath(__file__)) DATA_DIR = os.path.join(BASE_DIR, "data") LOG_DIR = os.path.join(BASE_DIR, "logs") if not os.path.exists(LOG_DIR): os.makedirs(LOG_DIR) LOG_FILE = os.path.join(LOG_DIR, "post.log") # ===== CSVパス ===== JA_CSV_PATH = os.path.join(DATA_DIR, "ja_posts.csv") MULTI_CSV_PATH = os.path.join(DATA_DIR, "multi_lang_posts.csv") # ===== X (Twitter) API 設定 ===== # 実際の値を環境変数または直書きで設定 X_BEARER_TOKEN_JA = "YOUR_JA_ACCOUNT_BEARER_TOKEN" X_BEARER_TOKEN_EN = "YOUR_EN_ACCOUNT_BEARER_TOKEN" X_BEARER_TOKEN_ZH = "YOUR_ZH_ACCOUNT_BEARER_TOKEN" X_TWEET_ENDPOINT = "https://api.twitter.com/2/tweets" # ===== Instagram Graph API 設定 ===== IG_USER_ID = "YOUR_IG_USER_ID" # InstagramビジネスアカウントID IG_ACCESS_TOKEN = "YOUR_IG_ACCESS_TOKEN" # 長期有効アクセストークン IG_API_BASE = "https://graph.facebook.com/v21.0" # あなたのドメインに変更 BASE_PUBLIC_URL = "https://your-domain.com/beautiful_future" IG_IMAGE_URLS = { "朝": [ f"{BASE_PUBLIC_URL}/public_images/morning/ig_morning_01_yellow_sunrise.png", f"{BASE_PUBLIC_URL}/public_images/morning/ig_morning_02_orange_morninglight.png", f"{BASE_PUBLIC_URL}/public_images/morning/ig_morning_03_skyblue_clear.png", f"{BASE_PUBLIC_URL}/public_images/morning/ig_morning_04_lightblue_fresh.png", f"{BASE_PUBLIC_URL}/public_images/morning/ig_morning_05_white_minimal.png", f"{BASE_PUBLIC_URL}/public_images/morning/ig_morning_06_pink_softdawn.png", f"{BASE_PUBLIC_URL}/public_images/morning/ig_morning_07_gold_horizon.png", f"{BASE_PUBLIC_URL}/public_images/morning/ig_morning_08_skyblue_cloud.png", f"{BASE_PUBLIC_URL}/public_images/morning/ig_morning_09_green_morningdew.png", f"{BASE_PUBLIC_URL}/public_images/morning/ig_morning_10_yellow_smile.png", ], "昼": [ f"{BASE_PUBLIC_URL}/public_images/noon/ig_noon_01_blue_sky.png", f"{BASE_PUBLIC_URL}/public_images/noon/ig_noon_02_cyan_energy.png", f"{BASE_PUBLIC_URL}/public_images/noon/ig_noon_03_white_clouds.png", f"{BASE_PUBLIC_URL}/public_images/noon/ig_noon_04_orange_active.png", f"{BASE_PUBLIC_URL}/public_images/noon/ig_noon_05_green_nature.png", f"{BASE_PUBLIC_URL}/public_images/noon/ig_noon_06_blue_momentum.png", f"{BASE_PUBLIC_URL}/public_images/noon/ig_noon_07_turquoise_refresh.png", f"{BASE_PUBLIC_URL}/public_images/noon/ig_noon_08_lightblue_clearair.png", f"{BASE_PUBLIC_URL}/public_images/noon/ig_noon_09_sunshine_bright.png", f"{BASE_PUBLIC_URL}/public_images/noon/ig_noon_10_yellow_focus.png", ], "夕方": [ f"{BASE_PUBLIC_URL}/public_images/night/ig_night_01_navy_calm.png", f"{BASE_PUBLIC_URL}/public_images/night/ig_night_02_darkblue_silent.png", f"{BASE_PUBLIC_URL}/public_images/night/ig_night_03_purple_relax.png", f"{BASE_PUBLIC_URL}/public_images/night/ig_night_04_black_minimal.png", f"{BASE_PUBLIC_URL}/public_images/night/ig_night_05_orange_sunset.png", f"{BASE_PUBLIC_URL}/public_images/night/ig_night_06_red_sunsetwarm.png", f"{BASE_PUBLIC_URL}/public_images/night/ig_night_07_darkgreen_deepforest.png", f"{BASE_PUBLIC_URL}/public_images/night/ig_night_08_blue_twilight.png", f"{BASE_PUBLIC_URL}/public_images/night/ig_night_09_gold_nightlight.png", f"{BASE_PUBLIC_URL}/public_images/night/ig_night_10_purple_moonlight.png", ], } # ===== 時刻テーブル (JST) ===== JA_TIME_TABLE = { "朝": (7, 30), "昼": (12, 15), "夕方": (20, 30), } EN_TIME_TABLE = { "Morning": (7, 30), "Noon": (12, 15), "Evening": (20, 30), }

# common_utils.py import json import csv import random from datetime import datetime import urllib.request import urllib.parse from config import ( JST, LOG_FILE, JA_CSV_PATH, MULTI_CSV_PATH, X_TWEET_ENDPOINT, IG_API_BASE, IG_ACCESS_TOKEN, IG_USER_ID, IG_IMAGE_URLS, ) def log(message: str): now = datetime.now(JST) line = f"[{now.isoformat()}] {message}" try: with open(LOG_FILE, "a", encoding="utf-8") as f: f.write(line + "\n") except Exception: print(line) def get_now_jst(): return datetime.now(JST) # ===== X API 共通 ===== def post_to_x(bearer_token: str, tweet_text: str) -> bool: if not bearer_token: log("ERROR: X bearer token not set") return False payload = {"text": tweet_text} data_bytes = json.dumps(payload).encode("utf-8") req = urllib.request.Request( X_TWEET_ENDPOINT, data=data_bytes, method="POST", headers={ "Authorization": f"Bearer {bearer_token}", "Content-Type": "application/json", } ) try: with urllib.request.urlopen(req) as resp: body = resp.read().decode("utf-8") log(f"X API response: {body}") return True except Exception as e: log(f"ERROR posting to X: {e}") return False # ===== Instagram API 共通 ===== def choose_random_image_url(slot: str): if slot not in IG_IMAGE_URLS: all_urls = [] for urls in IG_IMAGE_URLS.values(): all_urls.extend(urls) return random.choice(all_urls) if all_urls else None urls = IG_IMAGE_URLS[slot] return random.choice(urls) if urls else None def ig_create_media(image_url: str, caption: str): endpoint = f"{IG_API_BASE}/{IG_USER_ID}/media" params = { "image_url": image_url, "caption": caption, "access_token": IG_ACCESS_TOKEN, } data_bytes = urllib.parse.urlencode(params).encode("utf-8") req = urllib.request.Request(endpoint, data=data_bytes, method="POST") try: with urllib.request.urlopen(req) as resp: body = resp.read().decode("utf-8") log(f"IG create media response: {body}") res = json.loads(body) return res.get("id") except Exception as e: log(f"ERROR ig_create_media: {e}") return None def ig_publish_media(creation_id: str) -> bool: endpoint = f"{IG_API_BASE}/{IG_USER_ID}/media_publish" params = { "creation_id": creation_id, "access_token": IG_ACCESS_TOKEN, } data_bytes = urllib.parse.urlencode(params).encode("utf-8") req = urllib.request.Request(endpoint, data=data_bytes, method="POST") try: with urllib.request.urlopen(req) as resp: body = resp.read().decode("utf-8") log(f"IG publish media response: {body}") return True except Exception as e: log(f"ERROR ig_publish_media: {e}") return False def post_to_instagram(caption: str, slot: str) -> bool: if not IG_ACCESS_TOKEN or not IG_USER_ID: log("ERROR: IG_ACCESS_TOKEN or IG_USER_ID not set") return False image_url = choose_random_image_url(slot) if not image_url: log(f"IG: no image_url found for slot={slot}") return False log(f"IG: chosen image_url={image_url}") creation_id = ig_create_media(image_url, caption) if not creation_id: log("IG: creation_id is None, abort.") return False success = ig_publish_media(creation_id) return success # ===== CSV読み込み ===== def load_ja_post_for_today(slot: str, now: datetime): date_label = now.strftime("%m-%d") try: with open(JA_CSV_PATH, "r", encoding="utf-8") as f: reader = csv.DictReader(f) for row in reader: if row.get("date_label") == date_label and row.get("time_slot") == slot: text = row.get("text_ja", "").strip() tags = row.get("tags", "").strip() tweet_text = text if tags: tweet_text = f"{text} {tags}" return tweet_text except FileNotFoundError: log(f"JA_CSV_PATH not found: {JA_CSV_PATH}") return None log(f"No JA row found for date={date_label}, slot={slot}") return None def load_multi_post_for_today(slot: str, now: datetime, lang: str): date_label = now.strftime("%m-%d") key = "text_en" if lang == "en" else "text_cn" try: with open(MULTI_CSV_PATH, "r", encoding="utf-8") as f: reader = csv.DictReader(f) for row in reader: if row.get("date") == date_label and row.get("time_slot") == slot: text = row.get(key, "").strip() return text except FileNotFoundError: log(f"MULTI_CSV_PATH not found: {MULTI_CSV_PATH}") return None log(f"No {lang} row found for date={date_label}, slot={slot}") return None

#!/usr/local/bin/python3.7 # -*- coding: utf-8 -*- from config import JA_TIME_TABLE, X_BEARER_TOKEN_JA from common_utils import ( log, get_now_jst, post_to_x, post_to_instagram, load_ja_post_for_today, ) def detect_time_slot(now): h = now.hour m = now.minute for slot, (sh, sm) in JA_TIME_TABLE.items(): if h == sh and m == sm: return slot return None def main(): now = get_now_jst() slot = detect_time_slot(now) if slot is None: log("JA: No time slot matched (not 朝/昼/夕方). Exiting.") return log(f"JA: Detected slot={slot}") tweet_text = load_ja_post_for_today(slot, now) if not tweet_text: log("JA: No tweet_text found. Exiting.") return if len(tweet_text) > 270: log("JA: Tweet too long, truncating for X.") tweet_text_for_x = tweet_text[:270] + "…" else: tweet_text_for_x = tweet_text # X 投稿 log(f"JA: Posting to X: {tweet_text_for_x}") ok_x = post_to_x(X_BEARER_TOKEN_JA, tweet_text_for_x) if ok_x: log("JA: Tweet posted successfully.") else: log("JA: Tweet failed.") # Instagram 投稿（captionは全文） caption = tweet_text log(f"JA: Posting to Instagram (slot={slot})") ok_ig = post_to_instagram(caption, slot) if ok_ig: log("JA: Instagram post succeeded.") else: log("JA: Instagram post failed.") if __name__ == "__main__": main()

#!/usr/local/bin/python3.7 # -*- coding: utf-8 -*- from config import EN_TIME_TABLE, X_BEARER_TOKEN_EN from common_utils import ( log, get_now_jst, post_to_x, load_multi_post_for_today, ) def detect_time_slot(now): h = now.hour m = now.minute for slot, (sh, sm) in EN_TIME_TABLE.items(): if h == sh and m == sm: return slot return None def main(): now = get_now_jst() slot = detect_time_slot(now) if slot is None: log("EN: No time slot matched (not Morning/Noon/Evening). Exiting.") return log(f"EN: Detected slot={slot}") text = load_multi_post_for_today(slot, now, lang="en") if not text: log("EN: No text found. Exiting.") return tweet_text = f"{text} #BeautifulFuture #BeautifulMind" if len(tweet_text) > 270: log("EN: Tweet too long, truncating.") tweet_text = tweet_text[:270] + "…" log(f"EN: Posting to X: {tweet_text}") ok = post_to_x(X_BEARER_TOKEN_EN, tweet_text) if ok: log("EN: Tweet posted successfully.") else: log("EN: Tweet failed.") if __name__ == "__main__": main()

#!/usr/local/bin/python3.7 # -*- coding: utf-8 -*- from config import EN_TIME_TABLE, X_BEARER_TOKEN_ZH from common_utils import ( log, get_now_jst, post_to_x, load_multi_post_for_today, ) def detect_time_slot(now): h = now.hour m = now.minute for slot, (sh, sm) in EN_TIME_TABLE.items(): if h == sh and m == sm: return slot return None def main(): now = get_now_jst() slot = detect_time_slot(now) if slot is None: log("ZH: No time slot matched (not Morning/Noon/Evening). Exiting.") return log(f"ZH: Detected slot={slot}") text = load_multi_post_for_today(slot, now, lang="zh") if not text: log("ZH: No text found. Exiting.") return tweet_text = f"{text} #美好未来 #美好心念" if len(tweet_text) > 270: log("ZH: Tweet too long, truncating.") tweet_text = tweet_text[:270] + "…" log(f"ZH: Posting to X: {tweet_text}") ok = post_to_x(X_BEARER_TOKEN_ZH, tweet_text) if ok: log("ZH: Tweet posted successfully.") else: log("ZH: Tweet failed.") if __name__ == "__main__": main()