pipe_line_obsei.py

import logging
import json
from transformers import pipeline
from obsei_module.obsei.payload import TextPayload
from obsei_module.obsei.analyzer.classification_analyzer import ZeroShotClassificationAnalyzer, ClassificationAnalyzerConfig
from obsei_module.obsei.source.website_crawler_source import TrafilaturaCrawlerConfig, TrafilaturaCrawlerSource
from obsei_module.obsei.sink.http_sink import HttpSinkConfig, HttpSink
from obsei_module.obsei.analyzer.sentiment_analyzer import *
import connect_mongo
import pymongo


async def get_object_by_link(db_name, collection_name, link):
    collection = connect_mongo.connect_to_mongo(db_name, collection_name)
    
    if collection is None:
        print("Failed to connect to MongoDB.")
        return None

    result = collection.find_one({"link": link})
    
    # if result:
    #     print(f"Object found: {result}")
    # else:
    #     print(f"No object found for the link: {link}")
    
    return result


# Hàm kết nối và kiểm tra bản ghi trong DB khác
async def get_existing_record_by_name(db_name, collection_name, name, link):
    collection = connect_mongo.connect_to_mongo(db_name, collection_name)
    
    if collection is None:
        print("Failed to connect to MongoDB.")
        return None

    # Kiểm tra bản ghi có cùng name và link hay không
    result = collection.find_one({"name": name, "link": link})
    
    # if result:
    #     print(f"Existing record found for name: {name} and link: {link}")
    # else:
    #     print(f"No existing record found for name: {name} and link: {link}")
    
    return result

# Hàm lưu processed_text vào MongoDB với kiểm tra name
async def save_processed_text_to_mongo(db_name, collection_name, link, processed_text, name=None):
    collection = connect_mongo.connect_to_mongo(db_name, collection_name)

    if collection is None:
        print("Failed to connect to MongoDB.")
        return None

    # Tạo bản ghi mới hoặc cập nhật bản ghi cũ
    document = {
        "link": link,
        "processed_text": processed_text,
        "name": name,  # Thêm name cho bản ghi mới
    }

    

    # Kiểm tra sự tồn tại của bản ghi trong DB dự phòng
    existing_record = await get_existing_record_by_name(db_name, collection_name, name, link)
    
    if existing_record:
        # Nếu bản ghi tồn tại, cập nhật
        result = collection.update_one(
            {"_id": existing_record["_id"]},
            {
                "$set": document
            }
        )
        if result.modified_count > 0:
            print(f"Successfully updated the record for {link}.")
        else:
            print(f"No changes made to the record for {link}.")
    else:
        # Nếu bản ghi chưa tồn tại, tạo mới
        result = collection.insert_one(document)
        print(f"Successfully inserted a new record for {link}.")
    
    return result

# Hàm xử lý URL, lấy dữ liệu, phân tích và lưu processed_text vào MongoDB
async def process_url(url: str, db_name: str, collection_name: str,backup_db_name: str, backup_collection_name: str):
    """Crawl data from the URL and analyze it with a Zero-shot classification model."""
    logging.basicConfig(level=logging.INFO)
    logger = logging.getLogger(__name__)

    # Bước 1: Crawl dữ liệu từ URL
    crawler_config = TrafilaturaCrawlerConfig(
        urls=[url]
    )
    crawler = TrafilaturaCrawlerSource()

    crawled_data = crawler.lookup(config=crawler_config)
    if not crawled_data:
        logger.error("No data found from crawler")
        return {"error": "No data found from crawler"}
    

    # Bước 2: Cấu hình phân tích với Zero-shot classification
    analyzer_config = ClassificationAnalyzerConfig(
        labels=["Sports", "Politics", "Technology", "Entertainment"],
        multi_class_classification=False,
        add_positive_negative_labels=False
    )

    analyzer = ZeroShotClassificationAnalyzer(
        model_name_or_path="facebook/bart-large-mnli",
        device="auto"
    )

    # Phân tích dữ liệu crawled_data
    analysis_results = analyzer.analyze_input(
        source_response_list=crawled_data,
        analyzer_config=analyzer_config
    )
#     transformers_analyzer_config = TransformersSentimentAnalyzerConfig(
#     labels=["positive", "negative"],
#     multi_class_classification=False,
#     add_positive_negative_labels=True
# )
    # transformers_analyzer = TransformersSentimentAnalyzer(model_name_or_path="facebook/bart-large-mnli", device="auto")
    # transformers_results = transformers_analyzer.analyze_input(crawled_data, analyzer_config=transformers_analyzer_config)

    # Cấu hình HttpSink (nếu cần gửi dữ liệu đi nơi khác)
    http_sink_config = HttpSinkConfig(
        url="https://httpbin.org/post", 
        headers={"Content-type": "application/json"},
        base_payload={"common_field": "test cua VNY"},
    )

    http_sink = HttpSink()

    responses = http_sink.send_data(analysis_results, http_sink_config)

    response_data = []
    for i, response in enumerate(responses):
        response_content = response.read().decode("utf-8")
        response_json = json.loads(response_content)
        
        response_data.append({
            "response_index": i + 1,
            "content": response_json,
            "status_code": response.status,
            "headers": dict(response.getheaders())
        })

    content_data = [item["content"] for item in response_data]
    processed_text = content_data[0].get("json", {}).get('segmented_data', 'No processed text available.')
    processed_text1 = content_data[0].get("json", {})
    content_data = processed_text1.get('meta').get('text')
    existing_record = await get_object_by_link(db_name, collection_name, url)
    if existing_record:
     existing_segmented_data = existing_record.get("segmented_data", {})
     existing_segmented_data.update({"new_analysis": processed_text})
     await save_processed_text_to_mongo(backup_db_name, backup_collection_name, url, processed_text, name="Sentiment Analysis")
    else:
     await save_processed_text_to_mongo(backup_db_name, backup_collection_name, url, processed_text, name="Sentiment Analysis")

    return processed_text,content_data

# url = "https://laodong.vn/bong-da-quoc-te/vua-phat-goc-nicolas-jover-la-vo-gia-voi-arsenal-1431091.ldo"
# db_name = "test"
# import asyncio
# collection_name = "articles"
# backup_db_name = "backup_test"
# backup_collection_name = "articles_analysis"
# processed_text = asyncio.run(process_url(url, db_name, collection_name, backup_db_name, backup_collection_name))