copywriting_production/tests/mytest/models/t_odm_link.py

import asyncio
from pathlib import Path
from typing import List,Dict,Any

from beanie import WriteRules,PydanticObjectId
from config.settings import CFG
from src.manager.core.db import DbManager,AsinSeed
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 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

class AIPromptConfig(Document):
    """AI提示配置模型(嵌入式文档)"""
    name: str = Field(default=None, description="名称")
    introduce: Optional[str] = Field(default=None, description="介绍")
    template: str = Field(default=None, description="llamaindex 提示模板")
    keywords: Optional[Dict[Any, Any]] = Field(
        default={},
        description="llamaindex 模板中的变量"
    )
    owner: Optional[BackLink["UserConfig"]] = Field(
       original_field="ai_prompt_configs",
 
    )
    class Settings:
        name = "configs"  # MongoDB集合名称

class UserConfig(Document):
    user_name: str = Field(default=None, description="")
    email: Optional[str] = Field(default=None, description="")
    ai_prompt_configs:Optional[List[Link[AIPromptConfig]]] = Field(
        description="关联的AI提示配置"
    )
    created_at: datetime = Field(default_factory=datetime.now)
    updated_at: Optional[datetime] = Field(default=None)

    class Settings:
        name = "configs"  # MongoDB集合名称
        use_state_management = True  # 启用状态管理

    async def update_timestamp(self):
        """更新文档时间戳"""
        self.updated_at = datetime.now()
        await self.save()
async def create_user():
    # 初始化Beanie
    db_mongo = BaseMongoManager(db_name='test')
    await db_mongo.initialize()
    user_config = UserConfig(
        user_name="test_user",
        ai_prompt_configs=[
            AIPromptConfig(
                name="竞品和长尾词分析",
                introduce="没有变量",
                template="""
我是日本站的亚马逊运营，我在给我的产品选主要关键词和长尾关键词。

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

筛选长尾词的示例：
- 假设我的产品是电线保护，那么竞品关键词中，“隐藏排线管” 就不符合长尾词
- 假设我的产品是“防老化、防动物咬”用途，你就不能在竞品数据中选择不属于我这个使用场景的长尾关键词。
""", 
            )
        ]
    )
    # 谁使用了 Link 谁就要保存，并且传参 link_rule=WriteRules.WRITE
    await user_config.save(link_rule=WriteRules.WRITE)

async def add_product_prompt():
    # 初始化Beanie
    db_mongo = BaseMongoManager(db_name='test')
    await db_mongo.initialize()
    # 获取用户
    user_config = await UserConfig.find_one(UserConfig.user_name == "test_user", fetch_links=True)
    marketing = AIPromptConfig(
                name="营销文案",
                introduce="没有变量",
                template="""
我是日本站的亚马逊运营，帮我给我的产品分析出最佳文案。

请根据以上竞品数据，按以下规则分析：
- 结合日本市场特点分析
""", 
            )
    marketing.owner = user_config
    await marketing.save(link_rule=WriteRules.WRITE)
'''
❗ 使用外链和反向链接反而增加复杂度， MongoDB 一个文档可以很大，并不是非得外链接
缺点：
- 操作麻烦，每次都要先获取外链文档，再获取所属者，再按照顺序来保存
- 提取的时候要加 fetch_links 参数，否则无法直接访问对象
- 如果文档有反向链接，打印的时候会全部显示出来，无法直观观察数据
- 在 MongoDB GUI客户端中，无法直观地查看数据，各个文档特别分散
- 集合不清晰，各个文档散乱
'''
async def main():
    # await add_product_prompt()
    # return
    # 初始化Beanie
    db_mongo = BaseMongoManager(db_name='test')
    await db_mongo.initialize()
    # await create_user()
    # return 
    # 获取用户
    user_config = await UserConfig.find_one(UserConfig.user_name == "test_user")
    await user_config.fetch_link(UserConfig.ai_prompt_configs)
    # marketing_prompt = await AIPromptConfig.find_one(AIPromptConfig.name == "营销文案", fetch_links=True)
    # competitor_prompt = await AIPromptConfig.find_one(AIPromptConfig.name == "竞品和长尾词分析", fetch_links=True)
    # user_config.ai_prompt_configs = [marketing_prompt,competitor_prompt]
    # await user_config.save(link_rule=WriteRules.WRITE)
    # logger.info(f"{marketing_prompt.owner}")
    # logger.info(f"{competitor_prompt.owner}")
    logger.info(f"{user_config.ai_prompt_configs}")
    

if __name__ == "__main__":
    asyncio.run(main())