copywriting_production/tests/mytest/llamaindex_t/t_simple_mongo.py

# pip install llama-index-readers-mongodb
from llama_index.readers.mongodb import SimpleMongoReader
from config.settings import MONGO_URL, MONGO_DB_NAME
from src.manager.template_manager import TemplateManager, TemplateService


# Initialize SimpleMongoReader
reader = SimpleMongoReader(
    uri=MONGO_URL,  # Provide the URI if not using host and port
)
def simple_load_mongodata():
# Lazy load data from MongoDB
    documents = reader.load_data(
        db_name="test",  # Name of the database
        collection_name="Product",  # Name of the collection
        field_names=[
            "competitor_analyze"
        ],  # Names of the fields to concatenate (default: ["text"])
        separator="",  # Separator between fields (default: "")
        query_dict=None,  # Query to filter documents (default: None)
        max_docs=0,  # Maximum number of documents to load (default: 0)
        metadata_names=None,  # Names of the fields to add to metadata attribute (default: None)
    )
    for doc in documents:
        print(doc.get_content())

def get_jinja2_env():
    from jinja2 import Environment, meta

    env = Environment()
    template_source = '[{"$match": {"basic_info.name": "{{product_name}}"}}, {"$project": {"basic_info": 1, "_id": {{show_id}}}}]'
    parsed_content = env.parse(template_source)
    variables = meta.find_undeclared_variables(parsed_content)
    print(variables)  # 输出: {'product_name'}

async def query_dict_load_mongodata():
    manager = TemplateManager()
    await manager.initialize()
    tempalte_mmodel = await manager.get_template("product_info")
    manager.render_template(tempalte_mmodel, {"product_name": "测试"})
    await reader.aload_data(
        db_name="test",  # Name of the database
        collection_name="Product",  # Name of the collection 

    )

import asyncio
import aiofiles
import os
import sys

async def task():
    get_jinja2_env()
def main():
    asyncio.run(task())

if __name__ == "__main__":
    main()