Tùy biến Nâng cao

from vnstock_news.config.sitemap_resolver import DynamicSitemapResolver
from vnstock_news.config.sites import SITES_CONFIG

# 1. Khởi tạo resolver ở chế độ ghi log
resolver = DynamicSitemapResolver(debug=True)

# 2. Lấy cấu hình của website có sitemap động (ví dụ: Tiền Phong)
tienphong_config = SITES_CONFIG["tienphong"]

# 3. Tính toán URL sitemap phù hợp với thời điểm hiện tại (năm - tháng)
resolved_url = resolver.get_sitemap_url("tienphong", tienphong_config)
print(f"Sitemap URL được tạo: {resolved_url}")
# Kết quả trả về dạng: https://tienphong.vn/sitemaps/news-2026-10.xml

custom_advanced_selectors = {
    # Tiêu đề: Thử class "detail-title" trước, nếu không có chuyển sang thẻ meta
    "title_selector": [
        {"tag": "h1", "class": "detail-title"},
        {"tag": "meta", "attr": "property", "value": "og:title", "extract": "content"}
    ],
    
    # Sapo bài viết (phần tóm tắt ở đầu)
    "short_desc_selector": [
        {"tag": "h2", "class": "sapo-detail"},
        {"tag": "div", "class": "sapo"}
    ],
    
    # Thời gian xuất bản: Lấy thuộc tính datetime trong thẻ <time> hoặc content của meta
    "publish_time_selector": [
        {"tag": "time", "attr": "datetime"},
        {"tag": "meta", "attr": "property", "value": "article:published_time", "extract": "content"},
        {"tag": "span", "class": "post-time"}
    ],
    
    # Tác giả
    "author_selector": {"tag": "span", "class": "author-name"}
}

import asyncio
import pandas as pd
from vnstock_news import AsyncBatchCrawler

async def fetch_custom_unsupported_site():
    # 1. Định nghĩa cấu hình cho website mới
    custom_site_profile = {
        "site_name": "Trang tin tức custom",
        "domain": "trangtin.example",
        "config": {
            "title_selector": {"tag": "h1", "class": "detail-title"},
            "short_desc_selector": {"tag": "h2", "class": "detail-sapo"},
            "publish_time_selector": {"tag": "div", "class": "detail-time"},
            "author_selector": {"tag": "div", "class": "author-name"}
        }
    }

    # 2. Khởi tạo AsyncBatchCrawler với custom_config (không truyền site_name)
    client = AsyncBatchCrawler(
        custom_config=custom_site_profile, 
        max_concurrency=3, 
        debug=False
    )

    # 3. Chỉ định nguồn truy xuất (RSS Feed hoặc Sitemap XML)
    sources = ["https://trangtin.example/rss.xml"]

    # 4. Thực thi truy xuất dữ liệu
    df = await client.fetch_articles_async(
        sources=sources,
        top_n=5
    )

    if not df.empty:
        print(f"Truy xuất thành công {len(df)} bài viết từ nguồn tùy chỉnh:")
        print(df[["title", "publish_time", "author"]].head())
    else:
        print("Không lấy được dữ liệu.")

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

import asyncio
from vnstock_news.trending.analyzer import TrendingAnalyzer
from vnstock_news import AsyncBatchCrawler, SITES_CONFIG

async def analyze_market_trending_phrases():
    # 1. Tải bài viết từ CafeF
    site = "cafef"
    sitemap_url = SITES_CONFIG[site]["sitemap_url"]
    
    client = AsyncBatchCrawler(site_name=site, max_concurrency=3)
    df = await client.fetch_articles_async(sources=[sitemap_url], top_n=30)
    
    if df.empty:
        print("Không có dữ liệu bài viết để phân tích.")
        return
        
    # 2. Khởi tạo TrendingAnalyzer
    analyzer = TrendingAnalyzer(min_token_length=3)
    
    # 3. Cập nhật cụm từ (2, 3, 4 từ) từ tiêu đề và tóm tắt
    for _, row in df.iterrows():
        title = str(row.get("title", ""))
        desc = str(row.get("short_description", ""))
        full_text = f"{title} {desc}"
        
        analyzer.update_trends(full_text, ngram_range=[2, 3, 4])
        
    # 4. Lấy top 15 cụm từ xuất hiện nhiều nhất
    top_phrases = analyzer.get_top_trends(top_n=15)
    
    print("Danh sách cụm từ phổ biến:")
    for rank, (phrase, count) in enumerate(top_phrases.items(), 1):
        print(f" {rank:2d}. Cụm từ: '{phrase:25s}' - Số lần: {count:3d}")

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

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

warnings.filterwarnings("ignore")

async def run_multi_site_pipeline_and_health_check():
    health_reports = []
    collected = []
    
    # Liệt kê rõ từng trang thay vì lấy theo thứ tự danh mục, để chỉ chạy trên
    # các trang bạn đã kiểm tra robots.txt và điều kiện sử dụng.
    for site in ["cafef", "cafebiz", "vietstock", "vneconomy", "vnexpress"]:
        print(f"Đang kiểm tra trang: [{site.upper()}] ...")
        
        cfg = SITES_CONFIG[site]
        client = AsyncBatchCrawler(site_name=site, max_concurrency=2)
        
        sources = []
        if "rss" in cfg and cfg["rss"].get("urls"):
            sources = [cfg["rss"]["urls"][0]]
        elif "sitemap_url" in cfg:
            sources = [cfg["sitemap_url"]]
            
        if not sources:
            health_reports.append({"site": site, "status": "FAIL (No Source)"})
            continue

        try:
            df = await client.fetch_articles_async(sources=sources, top_n=3)
            
            if df.empty:
                health_reports.append({"site": site, "status": "EMPTY"})
            else:
                df["source_site"] = site
                collected.append(df)
                
                row = df.iloc[0]
                def _is_valid(val):
                    if pd.isna(val) or val is None: return False
                    if isinstance(val, str) and not val.strip(): return False
                    return True
                
                report = {
                    "site": site,
                    "status": "SUCCESS",
                    "title": "✅" if _is_valid(row.get("title")) else "❌",
                    "publish_time": "✅" if _is_valid(row.get("publish_time")) else "❌",
                    "author": "✅" if _is_valid(row.get("author")) else "❌"
                }
                health_reports.append(report)
                
        except Exception as e:
            health_reports.append({"site": site, "status": "ERROR", "error": str(e)})

    # Hiển thị kết quả kiểm tra
    df_health = pd.DataFrame(health_reports)
    print("\nKết quả kiểm tra dữ liệu:")
    print(df_health.to_string(index=False))

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