Hướng Dẫn Cào Dữ Liệu Python Telegram Bot Trong 15 Phút

pip install requests beautifulsoup4 python-dotenv

import requests
from bs4 import BeautifulSoup

def scrape_latest_deal():
    url = "https://toniblog.net/blog/so-sanh-gia-vps-nvme/"
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
    }

    try:
        response = requests.get(url, headers=headers, timeout=10)
        response.raise_for_status()

        # Parse HTML với BeautifulSoup
        soup = BeautifulSoup(response.text, "html.parser")
        
        # Lấy tiêu đề bài viết và thông tin deal
        title = soup.find("h1").text.strip()
        coupon_box = soup.find("blockquote")
        coupon_text = coupon_box.text.strip() if coupon_box else "Không tìm thấy mã"

        return {
            "title": title,
            "details": coupon_text,
            "url": url
        }
    except Exception as e:
        print(f"Lỗi cào dữ liệu: {e}")
        return None

if __name__ == "__main__":
    data = scrape_latest_deal()
    print("Dữ liệu cào được:", data)

pip install playwright
playwright install chromium

from playwright.sync_api import sync_playwright

def scrape_dynamic_price():
    url = "https://example.com/product-page"
    
    with sync_playwright() as p:
        # Khởi tạo trình duyệt Headless Chromium
        browser = p.chromium.launch(headless=True)
        page = browser.new_page()
        
        # Giả lập User-Agent trình duyệt thật
        page.set_extra_http_headers({
            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"
        })
        
        page.goto(url, wait_until="networkidle")
        
        # Chờ phần tử chứa giá sản phẩm xuất hiện trong DOM
        page.wait_for_selector(".product-price")
        
        price = page.locator(".product-price").inner_text()
        title = page.locator("h1.product-title").inner_text()
        
        browser.close()
        
        return {
            "title": title,
            "price": price,
            "url": url
        }

if __name__ == "__main__":
    print(scrape_dynamic_price())

import os
import requests
from dotenv import load_dotenv

load_dotenv()

BOT_TOKEN = os.getenv("BOT_TOKEN", "7890123456:AAFlx...your_token")
CHAT_ID = os.getenv("CHAT_ID", "123456789")

def send_telegram_alert(title, details, link):
    """
    Gửi tin nhắn định dạng HTML sang Telegram Bot
    """
    api_url = f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage"
    
    # Soạn thảo nội dung tin nhắn dạng HTML
    message_text = (
        f"🚀 <b>THÔNG BÁO TỰ ĐỘNG TỪ PYTHON BOT</b>\n\n"
        f"📌 <b>Chủ đề:</b> {title}\n"
        f"💡 <b>Chi tiết:</b> {details}\n\n"
        f"🔗 <a href='{link}'>Click vào đây để xem chi tiết</a>\n"
        f"⏰ <i>Thời gian: Vừa cập nhật</i>"
    )
    
    payload = {
        "chat_id": CHAT_ID,
        "text": message_text,
        "parse_mode": "HTML",
        "disable_web_page_preview": False
    }
    
    try:
        response = requests.post(api_url, json=payload, timeout=10)
        res_data = response.json()
        if res_data.get("ok"):
            print("✅ Đã gửi thông báo Telegram thành công!")
        else:
            print(f"❌ Lỗi gửi Telegram: {res_data.get('description')}")
    except Exception as e:
        print(f"❌ Lỗi kết nối Telegram API: {e}")

if __name__ == "__main__":
    send_telegram_alert(
        title="Ưu đãi VPS NVMe AZDIGI Giảm 30%",
        details="Mã giảm giá TONIDEV30 đã sẵn sàng sử dụng!",
        link="https://toniblog.net/blog/so-sanh-gia-vps-nvme/"
    )

pip install schedule

import time
import schedule
from scraper_static import scrape_latest_deal
from telegram_bot import send_telegram_alert

# Lưu trạng thái cũ để tránh gửi tin nhắn trùng lặp
last_scraped_title = None

def job():
    global last_scraped_title
    print("🔄 Đang tiến hành cào dữ liệu...")
    
    data = scrape_latest_deal()
    if not data:
        return
        
    # Chỉ gửi tin nhắn nếu có dữ liệu mới khác lần cào trước
    if data["title"] != last_scraped_title:
        last_scraped_title = data["title"]
        send_telegram_alert(
            title=data["title"],
            details=data["details"],
            link=data["url"]
        )
    else:
        print("ℹ️ Dữ liệu không có thay đổi mới. Không gửi tin nhắn.")

# Lập lịch chạy tự động mỗi 15 phút
schedule.every(15).minutes.do(job)

if __name__ == "__main__":
    print("🤖 Python Automation Bot đang chạy ngầm...")
    # Chạy lần đầu ngay khi bật script
    job()
    
    while True:
        schedule.run_pending()
        time.sleep(1)

# Sử dụng Python 3.11 Slim siêu nhẹ
FROM python:3.11-slim

WORKDIR /app

# Đặt biến môi trường Python không ghi cache file .pyc
ENV PYTHONUNBUFFERED=1 \
    PYTHONDONTWRITEBYTECODE=1

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["python", "main.py"]

version: '3.8'

services:
  python-telegram-scraper:
    build: .
    container_name: python_telegram_bot
    restart: always
    environment:
      - BOT_TOKEN=7890123456:AAFlx...your_token
      - CHAT_ID=123456789

docker compose up -d --build

docker logs -f python_telegram_bot