Upload 4 files

connect_mongo.py

@@ -0,0 +1,16 @@
+from pymongo import MongoClient
+from pymongo.server_api import ServerApi
+
+def connect_to_mongo(db_name: str, collection_name: str):
+
+    uri = "mongodb+srv://kltn20133118:XEYSVzYvEwsp5Cvo@cluster0.nnsw9.mongodb.net/?retryWrites=true&w=majority&appName=Cluster0"
+    try:
+     
+        client = MongoClient(uri, server_api=ServerApi('1'))       
+        db = client[db_name]
+        collection = db[collection_name]
+        return collection
+     
+    except Exception as e:
+        print("Lỗi kết nối MongoDB:", e)
+        return None

function.py

@@ -0,0 +1,136 @@
+from pymongo import MongoClient
+from typing import List, Any
+from connect_mongo import connect_to_mongo
+
+class NewsDatabaseHandler:
+    def __init__(self, uri: str, db_name: str):
+        """
+        Khởi tạo kết nối MongoDB một lần duy nhất.
+        """
+        self.client = MongoClient(uri)
+        self.db = self.client[db_name]
+
+    def get_all(self, collection_name: str) -> List[dict]:
+        """
+        Lấy tất cả bản ghi từ collection.
+        """
+        collection = self.db[collection_name]
+        return list(collection.find())
+
+    def find_by_category(self, collection_name: str, category: str) -> List[dict]:
+        """
+        Tìm các bản ghi theo category.
+        """
+        collection = self.db[collection_name]
+        query = {"category": category}
+        return list(collection.find(query))
+
+    def save_to_mongo(self, data: List[dict], collection_name: str):
+        """
+        Lưu dữ liệu vào MongoDB.
+        """
+        collection = self.db[collection_name]
+        collection.insert_many(data)
+
+    def close(self):
+        """
+        Đóng kết nối MongoDB.
+        """
+        self.client.close()
+
+from datetime import datetime
+
+def find_key():
+    db_name = "news"
+    collection_name = "articles"
+    
+    collection = connect_to_mongo(db_name, collection_name)
+    
+    # Tìm tất cả các bài viết có "name" chứa từ "Campuchia" và thuộc về TuoiTre hoặc VnExpress
+    vnexpress_or_tuoitre_news = collection.find({
+        "$in": [
+            {"VnExpress.name": {"$regex": "Campuchia", "$options": "i"}}  # Tìm kiếm trong VnExpres  # Tìm kiếm trong TuoiTre
+        ]
+    })
+
+    # In ra các kết quả tìm được
+    for result in vnexpress_or_tuoitre_news:
+        # Nếu bạn chỉ muốn thấy bài viết chứa từ "Campuchia" trong tên:
+        if "Campuchia" in result.get('VnExpress', [{}])[0].get('name', '') or "Campuchia" in result.get('TuoiTre', [{}])[0].get('name', ''):
+            print(result)
+
+# Gọi hàm tìm kiếm
+# find_key()
+
+    
+def get_vnexpress_news(category):
+    db_name = "news"
+    collection_name = "articles"
+    
+    collection = connect_to_mongo(db_name, collection_name)
+    
+    
+    # # Lấy thời gian hiện tại
+    # today = datetime.now().strftime("%Y-%m-%d")
+    # start_time = f"{today} 21:40:00"
+    # print(f"Start time: {start_time}")
+    
+    # # Truy vấn MongoDB
+    # vnexpress_news = collection.find({
+    #     "category": category,
+    #     "VnExpress.time": {"$gte": start_time}  # Điều kiện so sánh thời gian
+    # })
+
+ 
+
+    # filtered_news = []
+    # # Duyệt qua các bài báo và lọc ra các bài hợp lệ
+    # for result in vnexpress_news:
+    #     if "VnExpress" in result:
+    #         for news in result["VnExpress"]:
+    #             # Lọc bỏ các mục trống hoặc không có dữ liệu
+    #             if news and "time" in news:
+    #                 print(f"Checking time for news: {news['time']}")
+    #                 if news["time"] >= start_time:
+    #                     filtered_news.append(news)
+
+    # return filtered_news
+
+# Gọi hàm và in kết quả
+# news = get_vnexpress_news("VnExpress")
+
+
+# if not news:
+#     print("No news found matching the criteria.")
+# else:
+#     for article in news:
+#         print(f"Found article: {article}")
+# Cấu hình kết nối
+
+
+
+db_name = "test"
+collection_name = "articles"
+collection = connect_to_mongo(db_name, collection_name)
+
+import datetime
+
+
+# Tính thời gian 1h30 trước từ thời điểm hiện tại
+time_limit = datetime.datetime.now() - datetime.timedelta(minutes=20)
+
+time_limit = time_limit.replace(tzinfo=datetime.timezone.utc)
+print(time_limit)
+
+
+query = {
+    "category": "VnExpress",
+    "time": {"$gte": time_limit}
+}
+
+results = collection.find(query)
+
+for result in results:
+    print("Bản ghi:")
+    print(result)
+    print("\n")

main.py

@@ -0,0 +1,82 @@
+from fastapi import FastAPI, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel
+import pipe_line_obsei
+
+# Định nghĩa model request body
+class URLProcessRequest(BaseModel):
+    target_url: str  # URL cần phân tích
+    primary_db: str  # Tên database chính
+    primary_collection: str  # Tên collection chính
+    backup_db: str  # Tên database dự phòng
+    backup_collection: str  # Tên collection dự phòng
+
+# Khởi tạo FastAPI
+app = FastAPI(
+    title="ChatBot HCMUTE",
+    description="Python ChatBot is intended for use in the topic Customizing chatbots. With the construction of 2 students Vo Nhu Y - 20133118 and Nguyen Quang Phuc 20133080",
+    swagger_ui_parameters={"syntaxHighlight.theme": "obsidian"},
+    version="1.0.0",
+    contact={
+        "name": "Vo Nhu Y",
+        "url": "https://pychatbot1.streamlit.app",
+        "email": "vonhuy5112002@gmail.com",
+    },
+    license_info={
+        "name": "Apache 2.0",
+        "url": "https://www.apache.org/licenses/LICENSE-2.0.html",
+    }
+)
+
+# Cấu hình CORS
+origins = [
+    "http://localhost:8000",
+    "https://yourfrontendapp.com",  # Thêm domain của frontend nếu cần
+]
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=origins,
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+@app.get("/")
+async def root():
+    return {"message": "Welcome to ChatBot HCMUTE API!"}
+
+@app.post("/api/v1/obsei/process_url/")
+async def process_url_api(request: URLProcessRequest):
+    """
+    API nhận request body chứa thông tin URL và các thông tin database cần thiết,
+    sau đó xử lý URL, phân tích và lưu dữ liệu vào MongoDB.
+    """
+    try:
+        # Lấy dữ liệu từ request body
+        target_url = request.target_url
+        primary_db = request.primary_db
+        primary_collection = request.primary_collection
+        backup_db = request.backup_db
+        backup_collection = request.backup_collection
+
+        # Gọi hàm `process_url` đã định nghĩa
+        processed_text, content_data = await pipe_line_obsei.process_url(
+            target_url, primary_db, primary_collection, backup_db, backup_collection
+        )
+
+        return {
+            "processed_text": processed_text,
+            "content_data": content_data,
+        }
+    except Exception as e:
+        raise HTTPException(
+            status_code=500,
+            detail=f"An error occurred while processing the request: {str(e)}"
+        )
+
+
+# Chạy FastAPI server
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, port=8005)

pipe_line_obsei.py

@@ -0,0 +1,173 @@
+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))