copywriting_production/tests/mytest/t_move_s3_file.py

import asyncio
from pathlib import Path
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
from utils.logu import get_logger
from upath import UPath
from src.manager.manager_task import ManagerTask
logger = get_logger('test')
async def test_product_mongo():
    manager = ManagerTask()    
    await manager.db_mongo.initialize()
    product = await Product.find_one(Product.basic_info.name == "电线保护套")
    logger.info(f"{product}")
    asin_completed = manager.db.get_asin_completed()
    for asin_model in asin_completed:
        key = asin_model.asin
        competitor = product.competitor_crawl_data.get(key)
        logger.info(f"{asin_model.mhtml_path}")
        mthml_data = read_file(asin_model.mhtml_path)
        mhtml_file_name = UPath(asin_model.mhtml_path).name
        mhtml_without_ext = mhtml_file_name.split('.')[0]
        new_path = f's3://public/amazone/copywriting_production/output/asinseed/{mhtml_without_ext}/{mhtml_file_name}'
        res = save_to_file(mthml_data, new_path)
        logger.info(f"new path {res}")
        competitor.mhtml_path = res
    await product.save()
    return product
async def main():
   await test_product_mongo()

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