copywriting_production/tests/mytest/models/t_odm_promting_config.py

from src.manager.core.db_mongo import BaseMongoManager
from utils.file import save_to_file, read_file
from config.celery import app
# Remove direct task imports
from celery.result import AsyncResult
from src.models.product_model import (
    Product,CompetitorCrawlData,AICompetitorAnalyzeMainKeywords, 
    SearchAmazoneKeyResult, ProductBaseInfo, Variant,AICompetitorAnalyzeMainKeywordsResult,
    AIAnalyzeCompare
    )
from src.models.field_config import FieldConfig
from src.models.config_model import (UserConfig, AIPromptConfig)
from utils.logu import get_logger
from upath import UPath
logger = get_logger('test')
from datetime import datetime
from typing import Dict, List, Optional, Any
from pydantic import Field, BaseModel
from beanie import Document, BackLink, Link


import asyncio
import aiofiles
import os
import sys

def get_ai_prompt_config():
    ai_marketting_prompt_config = AIPromptConfig(
                name="营销文案",
                introduce="没有变量",
                template="""
我是日本站的亚马逊运营，帮我给我的产品分析出最佳文案。

请根据以上数据，按以下规则分析：
- 结合日本市场特点分析
""", 
    )
    ai_competitor_promting = AIPromptConfig(
                name="竞品和长尾词分析",
                introduce="没有变量",
                template="""
我是日本站的亚马逊运营，我在给我的产品选主要关键词和长尾关键词。

请根据以上竞品数据，按以下规则分析：
- 选出搜索量在1万以上的相同关键词作为主要关键词3个。
- 如果竞品的搜索量都不足1万，则从排名前十的关键词中筛选3个搜索量最大且相关性最强的词。
- 结合日本市场特点分析
- 根据我的产品基本信息，从竞品的主要信息和同类竞品的相似关键词中，筛选出最符合我产品的长尾关键词 tail_keys 12 个(如果存在的话)。

筛选长尾词的示例：
- 假设我的产品是电线保护，那么竞品关键词中，“隐藏排线管” 就不符合长尾词
- 假设我的产品是“防老化、防动物咬”用途，你就不能在竞品数据中选择不属于我这个使用场景的长尾关键词。
""", 
            )
    return ai_marketting_prompt_config,ai_competitor_promting
async def create_user():
    db_mongo = BaseMongoManager()
    await db_mongo.initialize()
    user = await UserConfig.find_one(UserConfig.user_name == "test_user")
    if user:
        return user
    user = UserConfig(
        user_name="test_user",
    )
    await user.save()

async def add_promt():
    db_mongo = BaseMongoManager()
    await db_mongo.initialize()
    await create_user()
    user = await UserConfig.find_one(UserConfig.user_name == "test_user")
    p = AIPromptConfig(
        name="竞品和长尾词分析-变量",
        template="""
我是日本站的亚马逊运营，我在给产品名称为 {product_name} 选主要关键词和长尾关键词。

请根据以上 {len_competitor_data} 个竞品数据，按以下规则分析：
- 选出搜索量在1万以上的相同关键词作为主要关键词 {main_key_num} 个。
- 如果竞品的搜索量都不足1万，则从排名前十的关键词中筛选 {main_key_num} 个搜索量最大且相关性最强的词。
- 结合日本市场特点分析
- 根据我的产品基本信息，从竞品的主要信息和同类竞品的相似关键词中，筛选出最符合我产品的长尾关键词 tail_keys {tail_key_num} 个以上

筛选长尾词的示例：
- 假设我的产品是电线保护，那么竞品关键词中，“隐藏排线管” 就不符合长尾词
- 假设我的产品是“防老化、防动物咬”用途，你就不能在竞品数据中选择不属于我这个使用场景的长尾关键词。
"""
    )
    p1,p2 = get_ai_prompt_config()
    await user.add_prompting(p)
    await user.add_prompting(p1)
    await user.add_prompting(p2)
# 主要是用于查找列表元素。但是返回字典，不够友好
async def find_demo():
    db_mongo = BaseMongoManager(db_name='test')
    await db_mongo.initialize()
    user = await UserConfig.find_one(UserConfig.user_name == "test_user")
    # 查找所有包含"营销文案"配置的用户
    pipeline = [
        {
            "$match": {
                "prompting.marketing.name": "营销文案"
            }
        },
        {
            "$project": {
                "results": {
                    "$filter": {
                        "input": "$prompting.marketing",
                        "as": "item",
                        "cond": { "$eq": ["$$item.name", "营销文案"] }
                    }
                }
            }
        }
    ]
    
    # 使用 aggregate 方法执行管道
    result = await UserConfig.aggregate(pipeline).to_list()
    logger.info(f"{result}")

    pipeline = [
        {
            "$match": {
                "prompting.competitor.name": "竞品和长尾词分析"
            }
        },
        {
            "$project": {
                "results": {
                    "$slice": [
                        {
                            "$filter": {
                                "input": "$prompting.competitor",
                                "as": "item",
                                "cond": { "$eq": ["$$item.name", "竞品和长尾词分析"] }
                            }
                        },
                        1
                    ]
                }
            }
        }
    ]
    
    # 使用正确的管道结构执行聚合查询
    result = await UserConfig.aggregate(pipeline).to_list()
    logger.info(f"{result}")
def main():
    asyncio.run(add_promt())

if __name__ == "__main__":
    main()