copywriting_production/tests/mytest/models/t_mongo_template_service.py

import asyncio
from src.models.template_model import Template, TemplateType
from src.manager.template_manager import TemplateManager, TemplateService
from src.manager.core.db_mongo import BaseMongoManager
from bson import json_util
import json
from utils.logu import get_logger

logger = get_logger('test', console_level='DEBUG')

import asyncio
import aiofiles
import os
import sys
product_info_pipeline = [
    {
        '$match': {
            'basic_info.name': '{{product_name}}'
        } 
    },
    {
        '$project': {
            'basic_info': 1, 
            '_id': 0
        }
    }
]
filter_competior_by_name = [
    {
        '$match': {
            'basic_info.name': "{{ product_name }}"
        }  
    },
    {
        '$project': {
            'competitor_crawl_data': 1
        }
    }, {
        '$addFields': {
            'competitors': {
                '$objectToArray': '$competitor_crawl_data'
            }
        }
    }, {
        '$unwind': '$competitors'
    }, {
        '$project': {
            '_id': 0, 
            'asin': '$competitors.k', 
            'product_info': {
                'main_text': '$competitors.v.extra_result.product_info.main_text'
            }, 
            'result_table': {
                '$map': {
                    'input': '$competitors.v.extra_result.result_table', 
                    'as': 'item', 
                    'in': {
                        'traffic_keyword': '$$item.traffic_keyword', 
                        'monthly_searches': '$$item.monthly_searches'
                    }
                }
            }
        }
    }
]
async def create_or_update_template():
    manager = TemplateManager()
    await manager.initialize()
    template = await manager.create_or_update_template(
        name="product_info",
        collection_name="Product",
        template_str=json.dumps(product_info_pipeline),
        template_type=TemplateType.AGGREGATION,
        description="产品信息查询模板"
    )
    await manager.create_or_update_template(
        name="competitor_for_llm",
        collection_name="Product",
        template_str=json.dumps(filter_competior_by_name),
        template_type=TemplateType.AGGREGATION,
        description="竞品数据查询模板，筛选出主要信息"
    )
    logger.info(f"Created template: {template}")

async def render():
    manager = TemplateManager()
    await manager.initialize()
    template = await manager.get_template("product_info")
    rendered_template = manager.render_template(template_str=template.template_str, context={"product_name": "电线保护套"})
    logger.info(f"product_info template: {rendered_template}")
    template = await manager.get_template("competitor_for_llm")
    rendered_template = manager.render_template(template_str=template.template_str, context={"product_name": "电线保护套"})
    logger.info(f"competitor_for_llm template: {rendered_template}")

async def task():
    manager = TemplateManager()
    await manager.initialize()
    result_list = await manager.execute_template(
        name="product_info",
        context={"product_name": "电线保护套"},
        collection_name="Product"
    )
    logger.info(f"product_info result: {result_list}")
    result_list = await manager.execute_template(
        name="competitor_for_llm",
        context={"product_name": "电线保护套"},
        collection_name="Product"
    )
    logger.info(f"competitor_for_llm result: {result_list}")
def main():
    asyncio.run(task())

if __name__ == "__main__":
    main()