bots

bot.py (6785B)

---

 import os
 import sys
 import subprocess
 import requests
 import frontmatter
 from bs4 import BeautifulSoup
 from telegram import Update
 from telegram.ext import ApplicationBuilder, CommandHandler, MessageHandler, filters, ContextTypes, ConversationHandler
 from dotenv import load_dotenv
 
 # --- CONFIGURATION ---
 load_dotenv()
 REPO_PATH = os.getenv("REPO_PATH") 
 MD_FILE_PATH = os.path.join(REPO_PATH, "content/listen.md")
 IMG_DIR = os.path.join(REPO_PATH, "content/assets/covers")
 MAX_SONGS = 9
 OWNER_ID = int(os.getenv("BOT_OWNER_ID", "0"))
 
 if OWNER_ID == 0:
     raise RuntimeError("BOT_OWNER_ID not set")
 
 # Conversation states
 TITLE, ARTIST, LINK, COVER = range(4)
 
 sys.stdout = open('/home/_listen-bot/bots/listen-page-bot/bot_debug.log', 'a', buffering=1)
 sys.stderr = sys.stdout
 
 print("--- Bot started ---")
 
 async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
     if not await is_authorized(update):
         await update.message.reply_text("Unauthorized.")
         return ConversationHandler.END
 
     await update.message.reply_text("Let's add a new song! What is the Song Title?")
     return TITLE
 
 async def get_title(update: Update, context: ContextTypes.DEFAULT_TYPE):
     if not await is_authorized(update):  
         await update.message.reply_text("Unauthorized.")
         return ConversationHandler.END
 
     context.user_data['title'] = update.message.text
     await update.message.reply_text(f"Artist for '{update.message.text}'?")
     return ARTIST
 
 async def get_artist(update: Update, context: ContextTypes.DEFAULT_TYPE):
     if not await is_authorized(update):
         await update.message.reply_text("Unauthorized.")
         return ConversationHandler.END
 
     context.user_data['artist'] = update.message.text
     await update.message.reply_text("Paste the YouTube Music link:")
     return LINK
 
 async def get_link(update: Update, context: ContextTypes.DEFAULT_TYPE):
     if not await is_authorized(update):
         await update.message.reply_text("Unauthorized.")
         return ConversationHandler.END
 
     context.user_data['link'] = update.message.text
     await update.message.reply_text("Finally, send the Album Cover image.")
     return COVER
 
 async def get_cover(update: Update, context: ContextTypes.DEFAULT_TYPE):
     if not await is_authorized(update):
         await update.message.reply_text("Unauthorized.")
         return ConversationHandler.END
 
     # Get the highest resolution photo
     photo_file = await update.message.photo[-1].get_file()
     
     # Generate filename (e.g., song_name.jpg)
     safe_name = "".join([c for c in context.user_data['title'] if c.isalnum()]).lower()
     img_filename = f"{safe_name}.jpg"
     img_path = os.path.join(IMG_DIR, img_filename)
     
     # Ensure directory exists
     os.makedirs(IMG_DIR, exist_ok=True)
     
     # Download image
     await photo_file.download_to_drive(img_path)
     
     # Update the Markdown file
     update_markdown(
         title=context.user_data['title'],
         artist=context.user_data['artist'],
         link=context.user_data['link'],
         img_url=f"/assets/covers/{img_filename}" # Relative path for HTML
     )
 
     run_local_deploy(REPO_PATH)
 
     await update.message.reply_text("Successfully updated listen.md and saved cover!")
     return ConversationHandler.END
 
 def update_markdown(title, artist, link, img_url):
     # Load the MD file
     post = frontmatter.load(MD_FILE_PATH)
     
     # Use BeautifulSoup to parse the HTML within the markdown content
     soup = BeautifulSoup(post.content, 'html.parser')
     gallery = soup.find('section', class_='song-gallery')
     
     if not gallery:
         # Fallback if section doesn't exist
         return
 
     # Create new song card HTML
     new_card_html = f'''
   <div class="song-card">
     <img src="{img_url}" alt="{title} Album Art">
     <div class="song-info">
       <span class="song-title">{title}</span>
       <span class="artist-name">{artist}</span>
       <a href="{link}" class="song-link">Listen</a>
     </div>
   </div>'''
     new_card_soup = BeautifulSoup(new_card_html, 'html.parser')
     
     # Add new card to the top (FIFO behavior)
     gallery.insert(0, new_card_soup)
     
     # Enforce MAX_SONGS (9)
     cards = gallery.find_all('div', class_='song-card')
     if len(cards) > MAX_SONGS:
         for extra in cards[MAX_SONGS:]:
             extra.decompose()
 
     # Format the content back (BeautifulSoup adds extra formatting, 
     # so we clean up the string slightly)
     post.content = f"\nAn overview of music and podcasts I currently have on repeat. [Checkout](blog/listen-cms/) how I manage this page to keep it current, without touching a single piece of code (until it breaks).\n\n## Music\n\n{gallery.prettify()}"
     
     # Save file
     with open(MD_FILE_PATH, 'wb') as f:
         frontmatter.dump(post, f)
 
 async def is_authorized(update):
     user = update.effective_user
     if not user:
         return False
 
     if user.id != OWNER_ID:
         print(f"Unauthorized access attempt from {user.id}")
         return False
 
     return True
 
 def run_local_deploy(repo_path):
     try:
         # 1. Git Add & Commit
         subprocess.run(["git", "add", "."], cwd=repo_path, check=True)
         subprocess.run(["git", "commit", "-m", "Bot: auto-update listen.md"], cwd=repo_path, check=True)
         
         # 2. Git Push 
         # (Works instantly if remote is a local path or SSH key has no passphrase)
         subprocess.run(["git", "push"], cwd=repo_path, check=True)
         
         # 3. Make Deploy
         # This calls the Makefile which uses 'doas rsync'
         subprocess.run(["make", "deploy"], cwd=repo_path, check=True)
         
         return True
     except subprocess.CalledProcessError as e:
         print(f"Deployment failed at: {e.cmd}")
         return False
 
 async def cancel(update: Update, context: ContextTypes.DEFAULT_TYPE):
     await update.message.reply_text("Cancelled.")
     return ConversationHandler.END
 
 if __name__ == '__main__':
     repo_path = REPO_PATH
     subprocess.run(["git", "fetch", "origin" ], cwd=repo_path, check=True)
     subprocess.run(["git", "pull" ], cwd=repo_path, check=True)
 
     app = ApplicationBuilder().token(os.getenv("TELEGRAM_TOKEN")).build()
 
     conv_handler = ConversationHandler(
         entry_points=[CommandHandler('add', start)],
         states={
             TITLE: [MessageHandler(filters.TEXT & ~filters.COMMAND, get_title)],
             ARTIST: [MessageHandler(filters.TEXT & ~filters.COMMAND, get_artist)],
             LINK: [MessageHandler(filters.TEXT & ~filters.COMMAND, get_link)],
             COVER: [MessageHandler(filters.PHOTO, get_cover)],
         },
         fallbacks=[CommandHandler('cancel', cancel)],
     )
 
     app.add_handler(conv_handler)
     print("Bot is running...")
     app.run_polling()