Ví dụ Thực tế

import time
import schedule
import pandas as pd
from vnstock_news import Crawler

def monitor_realtime_financial_news():
    print("Đang tiến hành quét các bản tin mới nhất qua RSS...")
    
    # 1. Khởi tạo Crawler với nguồn báo hỗ trợ RSS (ví dụ: Dân Trí, CafeBiz, VnExpress)
    crawler = Crawler(site_name="dantri")
    
    # 2. Lấy 10 bài viết mới nhất
    articles = crawler.get_articles_from_feed(limit_per_feed=10)
    
    if articles:
        df = pd.DataFrame(articles)
        print(f"Ghi nhận {len(df)} bài viết mới:")
        for idx, row in df.head(3).iterrows():
            print(f" - [{row.get('publish_time')}] {row.get('title')} ({row.get('url')})")
    else:
        print("Không có bản tin mới trong lượt quét này.")

# Lên lịch tự động chạy mỗi 15 phút một lần
schedule.every(15).minutes.do(monitor_realtime_financial_news)

if __name__ == "__main__":
    # Chạy lần đầu
    monitor_realtime_financial_news()
    
    # Duyệt vòng lặp chờ theo lịch
    print("Chương trình theo dõi tin tức đang chạy...")
    while True:
        schedule.run_pending()
        time.sleep(10)

import os
import pandas as pd
from vnstock_news import Crawler

def fetch_combined_rss_and_sitemap_news():
    site = "cafebiz"
    crawler = Crawler(site_name=site)
    
    # Bước 1: Lấy bài viết từ RSS Feed
    print("1. Đang quét RSS Feed...")
    rss_articles = crawler.get_articles_from_feed(limit_per_feed=20)
    rss_df = pd.DataFrame(rss_articles) if rss_articles else pd.DataFrame()
    
    # Bước 2: Lấy bài viết từ Sitemap XML
    print("2. Đang quét Sitemap XML...")
    sitemap_articles = crawler.get_articles(limit=50)
    sitemap_df = pd.DataFrame(sitemap_articles) if sitemap_articles else pd.DataFrame()
    
    # Bước 3: Gộp dữ liệu và loại bỏ các bài viết trùng URL
    if not rss_df.empty or not sitemap_df.empty:
        if not sitemap_df.empty and 'loc' in sitemap_df.columns:
            sitemap_df = sitemap_df.rename(columns={'loc': 'url'})
            
        combined_df = pd.concat([rss_df, sitemap_df], ignore_index=True)
        
        total_before = len(combined_df)
        combined_df = combined_df.drop_duplicates(subset=['url'])
        total_after = len(combined_df)
        
        os.makedirs("output", exist_ok=True)
        out_csv = "output/combined_market_news.csv"
        combined_df.to_csv(out_csv, index=False, encoding="utf-8-sig")
        
        print(f"Hoàn tất gộp. Loại bỏ {total_before - total_after} bài trùng.")
        print(f"Tổng số bài viết thu được: {total_after}")
        print(f"Dữ liệu lưu tại: {out_csv}")

if __name__ == "__main__":
    fetch_combined_rss_and_sitemap_news()

import os
import asyncio
import pandas as pd
from vnstock_news import AsyncBatchCrawler, SITES_CONFIG

async def generate_nlp_training_dataset():
    target_publisher = "cafef"
    config = SITES_CONFIG[target_publisher]
    sitemap_url = config.get("sitemap_url") or config.get("sitemap", {}).get("current_url")
    
    if not sitemap_url:
        print("Cấu hình Sitemap không tồn tại.")
        return

    # Khởi tạo AsyncBatchCrawler với 3 luồng kết nối song song
    crawler = AsyncBatchCrawler(site_name=target_publisher, max_concurrency=3)
    
    print(f"Đang thu thập bài viết từ sitemap: {sitemap_url} ...")
    
    articles_df = await crawler.fetch_articles_async(
        sources=[sitemap_url],
        top_n=50,       # Lấy 50 bài viết
        within="365d"   # Lọc bài viết trong 1 năm
    )
    
    if not articles_df.empty:
        os.makedirs("output", exist_ok=True)
        csv_file = "output/nlp_financial_dataset.csv"
        articles_df.to_csv(csv_file, index=False, encoding="utf-8-sig")
        
        print(f"Hoàn tất tải {len(articles_df)} bài viết.")
        print(articles_df[['title', 'publish_time', 'author']].head(3))
        
        sample_md = articles_df.iloc[0].get('content', '')
        print(f"Độ dài nội dung bài viết mẫu: {len(sample_md)} ký tự.")
        print(f"Dữ liệu lưu tại: {csv_file}")
    else:
        print("Không thu thập được dữ liệu.")

if __name__ == "__main__":
    asyncio.run(generate_nlp_training_dataset())

import re
import pandas as pd
from collections import Counter

def clean_vietnamese_text(text):
    """Xóa thẻ HTML và khoảng trắng thừa"""
    if pd.isna(text) or not text:
        return ""
    text = re.sub(r'<[^>]+>', '', str(text))
    text = re.sub(r'[\r\n\t]+', ' ', text)
    text = re.sub(r'\s+', ' ', text).strip()
    return text

def normalize_dataframe_timestamps(df, datetime_col="publish_time"):
    """Chuẩn hóa thời gian về múi giờ Việt Nam (UTC+7)"""
    if datetime_col in df.columns:
        df[datetime_col] = pd.to_datetime(df[datetime_col], errors='coerce')
        if df[datetime_col].dt.tz is None:
            df[datetime_col] = df[datetime_col].dt.tz_localize('UTC')
        df[datetime_col] = df[datetime_col].dt.tz_convert('Asia/Ho_Chi_Minh')
    return df

def extract_top_keywords(df, text_column="title", top_n=15):
    """Thống kê các từ xuất hiện nhiều nhất trong tiêu đề"""
    words = []
    for text in df[text_column].dropna():
        cleaned = clean_vietnamese_text(text).lower()
        tokens = [w for w in re.findall(r'\w+', cleaned) if len(w) >= 3]
        words.extend(tokens)
        
    counts = Counter(words)
    return counts.most_common(top_n)

if __name__ == "__main__":
    raw_data = [
        {
            "title": "Cổ phiếu Vietcombank (VCB) tăng vọt vượt đỉnh lịch sử",
            "content": "<p>Thị trường chứng khoán chứng kiến <b>lực cầu mạnh</b>...</p>",
            "publish_time": "2026-08-07T02:30:00Z"
        },
        {
            "title": "Ngân hàng Nhà nước giảm lãi suất điều hành hỗ trợ doanh nghiệp",
            "content": "<div>Động thái hạ lãi suất giúp tăng thanh khoản...</div>",
            "publish_time": "2026-08-07T04:15:00Z"
        }
    ]
    
    df = pd.DataFrame(raw_data)
    
    # 1. Làm sạch văn bản
    df['title'] = df['title'].apply(clean_vietnamese_text)
    df['content'] = df['content'].apply(clean_vietnamese_text)
    
    # 2. Chuẩn hóa múi giờ
    df = normalize_dataframe_timestamps(df)
    
    print("Dữ liệu sau khi chuẩn hóa múi giờ:")
    print(df[['title', 'publish_time']])
    
    # 3. Trích xuất từ khóa
    top_words = extract_top_keywords(df, text_column="title")
    print("\nDanh sách từ khóa phổ biến:")
    for word, freq in top_words:
        print(f" - {word:15s}: {freq} lần")

import sqlite3
import pandas as pd
from vnstock_news import Crawler

def init_news_database(db_path="financial_news.db"):
    """Tạo bảng cơ sở dữ liệu và chỉ mục"""
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS articles (
            url TEXT PRIMARY KEY,
            title TEXT NOT NULL,
            short_description TEXT,
            content TEXT,
            publish_time TEXT,
            author TEXT,
            source TEXT,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """)
    cursor.execute("CREATE INDEX IF NOT EXISTS idx_publish_time ON articles(publish_time)")
    cursor.execute("CREATE INDEX IF NOT EXISTS idx_source ON articles(source)")
    
    conn.commit()
    conn.close()

def upsert_articles_to_db(articles_list, source_name="dantri", db_path="financial_news.db"):
    """Ghi dữ liệu vào SQLite"""
    if not articles_list:
        return 0
        
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    
    upsert_query = """
        INSERT OR REPLACE INTO articles 
        (url, title, short_description, content, publish_time, author, source)
        VALUES (?, ?, ?, ?, ?, ?, ?)
    """
    
    records = []
    for a in articles_list:
        records.append((
            a.get("url"),
            a.get("title"),
            a.get("short_description"),
            a.get("content"),
            str(a.get("publish_time")),
            a.get("author"),
            source_name
        ))
        
    cursor.executemany(upsert_query, records)
    conn.commit()
    inserted_count = cursor.rowcount
    conn.close()
    
    return inserted_count

if __name__ == "__main__":
    db_file = "financial_news.db"
    init_news_database(db_file)
    
    crawler = Crawler(site_name="dantri")
    articles = crawler.get_articles_from_feed(limit_per_feed=15)
    
    if articles:
        count = upsert_articles_to_db(articles, source_name="dantri", db_path=db_file)
        print(f"Đã chèn hoặc cập nhật {count} bản ghi vào CSDL SQLite: {db_file}")
        
        conn = sqlite3.connect(db_file)
        df_db = pd.read_sql_query("SELECT url, title, publish_time, source FROM articles LIMIT 5", conn)
        conn.close()
        print("\nMẫu dữ liệu trong CSDL:")
        print(df_db)

import os
import time
import logging
import pandas as pd
from vnstock_news import BatchCrawler

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler("crawler_pipeline.log", encoding="utf-8"),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger("VNStockNewsPipeline")

PROGRESS_FILE = "downloaded_urls.txt"

def load_downloaded_urls():
    """Đọc danh sách các URL đã tải thành công"""
    if os.path.exists(PROGRESS_FILE):
        with open(PROGRESS_FILE, "r", encoding="utf-8") as f:
            return set(line.strip() for line in f if line.strip())
    return set()

def save_downloaded_url(url):
    """Lưu URL vừa thu thập vào nhật ký tiến độ"""
    with open(PROGRESS_FILE, "a", encoding="utf-8") as f:
        f.write(f"{url}\n")

def run_safe_resumable_pipeline():
    logger.info("Bắt đầu tiến trình thu thập tin tức...")
    
    downloaded_urls = load_downloaded_urls()
    logger.info(f"Đã tải {len(downloaded_urls)} URL từ tiến độ trước đó.")
    
    crawler = BatchCrawler(
        site_name="cafef",
        request_delay=1.5,
        output_path="./production_output"
    )
    
    try:
        articles_df = crawler.fetch_articles(limit=30)
        
        if articles_df.empty:
            logger.warning("Không tìm thấy bài viết mới.")
            return

        new_records = []
        for idx, row in articles_df.iterrows():
            url = row.get("url")
            if not url or url in downloaded_urls:
                continue
                
            new_records.append(row)
            save_downloaded_url(url)
            
        if new_records:
            df_new = pd.DataFrame(new_records)
            timestamp = time.strftime("%Y%m%d_%H%M%S")
            out_path = f"./production_output/news_batch_{timestamp}.csv"
            os.makedirs("./production_output", exist_ok=True)
            df_new.to_csv(out_path, index=False, encoding="utf-8-sig")
            logger.info(f"Đã thu thập thêm {len(df_new)} bài viết mới. Lưu file: {out_path}")
        else:
            logger.info("Tất cả bài viết đã được lưu từ trước.")
            
    except Exception as e:
        if "429" in str(e) or "Too Many Requests" in str(e):
            logger.error("Phát hiện giới hạn truy cập (Rate Limit 429). Tạm dừng 15 phút...")
            time.sleep(900)
        else:
            logger.error(f"Phát sinh lỗi: {e}")

if __name__ == "__main__":
    run_safe_resumable_pipeline()

# Mở trình chỉnh sửa Crontab
crontab -e

# Thêm dòng sau vào file crontab:
0 8,18 * * * /usr/bin/python3 /path/to/production_pipeline.py >> /path/to/cron.log 2>&1