import asyncio
import logging
from datetime import datetime, timezone, timedelta

from sqlalchemy import select, delete
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.base import async_session
from app.models.material_cost import MaterialCost
from app.models.user_oauth import UserOAuth
from app.models.user_oauth_account import UserOAuthAccount
from app.models.resources_material import ResourcesMaterial
from app.utils.id_gen import generate_id
from app.utils.douyinApi import DouyinApi

import os
import json

LOG_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "logs")
os.makedirs(LOG_DIR, exist_ok=True)

logger = logging.getLogger("material_consumption_task")
logger.setLevel(logging.INFO)


class DailyRotatingFileHandler(logging.FileHandler):
    def __init__(self, directory, encoding=None):
        self.directory = directory
        filename = self._get_log_filename()
        super().__init__(filename, encoding=encoding)
    
    def _get_log_filename(self):
        return os.path.join(self.directory, f"material_consumption-{datetime.now(timezone.utc).strftime('%Y-%m-%d')}.log")
    
    def emit(self, record):
        current_filename = self._get_log_filename()
        if self.baseFilename != current_filename:
            self.close()
            self.baseFilename = current_filename
            self.stream = self._open()
        super().emit(record)


if not logger.handlers:
    handler = DailyRotatingFileHandler(LOG_DIR, encoding="utf-8")
    handler.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(message)s", "%Y-%m-%d %H:%M:%S"))
    logger.addHandler(handler)

douyin_api = DouyinApi()


def _convert_numeric(value):
    """Convert string numeric values to appropriate Python types."""
    if value is None or value == "":
        return None
    try:
        # Try int first
        return int(value)
    except ValueError:
        try:
            # Then try float
            return float(value)
        except ValueError:
            # Return original value if conversion fails
            return None


class MaterialConsumptionQueue:
    def __init__(self):
        self.queue: asyncio.Queue[dict] = asyncio.Queue()
        self.running = False

    async def enqueue(self, task_data: dict):
        """Add a task to the queue."""
        await self.queue.put(task_data)

    async def run(self):
        """Main processing loop."""
        self.running = True
        logger.info("Material consumption queue started")

        while self.running:
            try:
                task_data = await asyncio.wait_for(self.queue.get(), timeout=5.0)
            except asyncio.TimeoutError:
                continue

            try:
                await self._process(task_data)
            except Exception as e:
                logger.error(f"Error processing material consumption task: {e}")
            finally:
                self.queue.task_done()

        logger.info("Material consumption queue stopped")

    async def _process(self, task_data: dict):
        """Process a single consumption update task."""
        oauth_id = task_data.get("oauth_id")
        advertiser_id = task_data.get("advertiser_id")
        date = task_data.get("date")
        
        logger.info(f"Processing consumption update for oauth_id={oauth_id}, advertiser_id={advertiser_id}, date={date}")
        
        try:
            result = await _fetch_and_save_consumption(oauth_id, advertiser_id, date)
            if result.get("success"):
                logger.info(f"Successfully updated consumption for advertiser {advertiser_id} on {date}")
            else:
                logger.error(f"Failed to update consumption for advertiser {advertiser_id} on {date}: {result.get('error')}")
        except Exception as e:
            logger.error(f"Exception processing consumption for advertiser {advertiser_id} on {date}: {e}")

    def stop(self):
        """Stop the queue."""
        self.running = False


async def _fetch_and_save_consumption(oauth_id: str, advertiser_id: str, date: str) -> dict:
    async with async_session() as db:
        try:
            consume_date_obj = datetime.strptime(date, "%Y-%m-%d").date()
            
            # 查询当前授权下系统上传的素材ID列表
            material_result = await db.execute(
                select(ResourcesMaterial.upload_id)
                .where(
                    ResourcesMaterial.oauth_id == oauth_id,
                    ResourcesMaterial.advertiser_id == advertiser_id,
                    ResourcesMaterial.deleted_at.is_(None),
                    ResourcesMaterial.upload_id.isnot(None),
                    ResourcesMaterial.upload_id != "",
                )
            )
            system_material_ids = [str(row[0]) for row in material_result.all()]
            
            if not system_material_ids:
                return {
                    "success": True,
                    "message": "没有系统上传的素材"
                }
            
            #强制删除旧数据,真实删除，不要软删除
            await db.execute(
                delete(MaterialCost).where(
                    MaterialCost.oauth_id == oauth_id,
                    MaterialCost.advertiser_id == advertiser_id,
                    MaterialCost.consume_date == consume_date_obj,
                    MaterialCost.deleted_at.is_(None),
                )
            )
            
            BATCH_SIZE = 50
            total_saved = 0
            total_requests = 0
            
            # 分批请求，每次请求50个素材
            for i in range(0, len(system_material_ids), BATCH_SIZE):
                batch_materials = system_material_ids[i:i+BATCH_SIZE]
                total_requests += 1
                
                params = {
                    "dimensions": json.dumps(["ad_platform_material_name","image_mode","material_id","stat_time_day"]),
                    "advertiser_id": int(advertiser_id),
                    "metrics": json.dumps([
                        "stat_cost","show_cnt","cpm_platform","click_cnt","ctr","cpc_platform","convert_cnt","conversion_cost","conversion_rate","deep_convert_cnt","deep_convert_cost","deep_convert_rate","click_start_cnt","click_start_cost","download_finish_rate",
                        "install_finish_cnt","install_finish_cost","install_finish_rate","active","active_cost",
                        "active_rate","active_register","active_register_cost","active_register_rate","game_addiction",
                        "game_addiction_cost","game_addiction_rate","attribution_next_day_open_cnt","attribution_next_day_open_cost","attribution_next_day_open_rate","next_day_open","active_pay","active_pay_cost","active_pay_rate",
                        "game_pay_count","game_pay_cost","in_app_uv","in_app_detail_uv","in_app_cart","in_app_pay",
                        "in_app_order","attribution_billing_game_in_app_ltv_1day","attribution_billing_game_in_app_roi_1day","phone","form",
                        "form_submit","map","button","view","download_start","qq","vote","lottery","message",
                        "redirect","shopping","consult","consult_effective","phone_confirm","phone_connect","phone_effective",
                        "redirect_to_shop","coupon_single_page","poi_address_click","poi_collect","customer_effective",
                        "attribution_customer_effective","attribution_customer_effective_cost","attribution_clue_pay_succeed",
                        "attribution_clue_pay_succeed_cost","attribution_clue_interflow","attribution_clue_interflow_cost","attribution_clue_high_intention","attribution_clue_high_intention_cost",
                        "consult_clue","clue_message_count","attribution_work_wechat_unfriend_count","attribution_clue_connected_count","attribution_clue_connected_cost","attribution_clue_connected_rate","attribution_form","form_and_submit_count","form_and_submit_cost","intention_form_and_submit_count","intention_form_and_submit_cost","attribution_micro_game_0d_ltv","attribution_micro_game_3d_ltv","attribution_micro_game_7d_ltv","attribution_micro_game_0d_roi","attribution_micro_game_3d_roi","attribution_micro_game_7d_roi","attribution_game_in_app_ltv_1day","attribution_game_in_app_roi_1day","active_pay_intra_day_count","active_pay_intra_day_cost","active_pay_intra_day_rate","first_pay_intra_24hour_amount","loan_completion","loan_completion_cost","loan_completion_rate","pre_loan_credit","pre_loan_credit_cost","loan_credit_cost","loan_credit","loan_credit_rate","in_wechat_pay_count","unfollow_in_wechat_count","loan","loan_cost","loan_rate","open_account_count","withdraw_m2_count","in_app_order_gmv","in_app_order_roi","in_app_pay_gmv","in_app_pay_roi","first_rental_order_count","commute_first_pay_count","first_order_count","submit_certification_count","approval_count","total_play","play_duration_3s","valid_play","valid_play_cost","valid_play_rate","valid_play_of_mille","valid_play_cost_of_mille","average_play_time_per_play","play_over_rate","dy_like","dy_comment","dy_share","report_cnt","location_click","dislike_cnt","dy_home_visited","dy_follow","message_action","click_landing_page","click_shopwindow","click_website","click_call_dy","click_download","luban_live_enter_cnt","luban_live_follow_cnt","luban_live_share_cnt","luban_live_comment_cnt","luban_live_gift_cnt","luban_live_gift_amount","click_call_cnt","click_counsel","a3_ask_count","a3_ask_cost"
                    ]),
                    "filters": json.dumps([
                        {
                            "field": "stat_cost",
                            "type": 3,
                            "operator": 4,
                            "values": ["0"]
                        },
                        {
                            "field": "material_id",
                            "type": 2,
                            "operator": 7,
                            "values": batch_materials
                        }
                    ]),
                    "start_time": date,
                    "end_time": date,
                    "order_by": json.dumps([
                        {
                            "field": "stat_cost",
                            "type": "ASC"
                        }
                    ])
                }

                response = await douyin_api.get_material_cost(oauth_id, params, request_count=3)
                
                if response.get("code") != 0:
                    logger.warning(f"获取素材消耗失败(批次 {i//BATCH_SIZE + 1}): {response.get('message', '未知错误')}")
                    continue
                
                data = response.get("data", {})
                list_data = data.get("rows", [])
                
                for item in list_data:
                    dimensions = item.get("dimensions", {})
                    metrics = item.get("metrics", {})
                    material_id = dimensions.get("material_id", "")
                    
                    cost = MaterialCost(
                        id=generate_id(),
                        oauth_id=oauth_id,
                        advertiser_id=str(advertiser_id),
                        material_id=material_id,
                        consume_date=consume_date_obj,
                        stat_cost=_convert_numeric(metrics.get("stat_cost")),
                        show_cnt=_convert_numeric(metrics.get("show_cnt")),
                        cpm_platform=_convert_numeric(metrics.get("cpm_platform")),
                        click_cnt=_convert_numeric(metrics.get("click_cnt")),
                        ctr=_convert_numeric(metrics.get("ctr")),
                        cpc_platform=_convert_numeric(metrics.get("cpc_platform")),
                        convert_cnt=_convert_numeric(metrics.get("convert_cnt")),
                        conversion_cost=_convert_numeric(metrics.get("conversion_cost")),
                        conversion_rate=_convert_numeric(metrics.get("conversion_rate")),
                        deep_convert_cnt=_convert_numeric(metrics.get("deep_convert_cnt")),
                        deep_convert_cost=_convert_numeric(metrics.get("deep_convert_cost")),
                        deep_convert_rate=_convert_numeric(metrics.get("deep_convert_rate")),
                        active=_convert_numeric(metrics.get("active")),
                        active_cost=_convert_numeric(metrics.get("active_cost")),
                        active_rate=_convert_numeric(metrics.get("active_rate")),
                        active_register=_convert_numeric(metrics.get("active_register")),
                        active_register_cost=_convert_numeric(metrics.get("active_register_cost")),
                        active_register_rate=_convert_numeric(metrics.get("active_register_rate")),
                        attribution_next_day_open_cnt=_convert_numeric(metrics.get("attribution_next_day_open_cnt")),
                        attribution_next_day_open_cost=_convert_numeric(metrics.get("attribution_next_day_open_cost")),
                        attribution_next_day_open_rate=_convert_numeric(metrics.get("attribution_next_day_open_rate")),
                        active_pay=_convert_numeric(metrics.get("active_pay")),
                        active_pay_cost=_convert_numeric(metrics.get("active_pay_cost")),
                        active_pay_rate=_convert_numeric(metrics.get("active_pay_rate")),
                        phone=_convert_numeric(metrics.get("phone")),
                        form=_convert_numeric(metrics.get("form")),
                        download_start=_convert_numeric(metrics.get("download_start")),
                        form_submit=_convert_numeric(metrics.get("form_submit")),
                        button=_convert_numeric(metrics.get("button")),
                        view=_convert_numeric(metrics.get("view")),
                        message=_convert_numeric(metrics.get("message")),
                        consult=_convert_numeric(metrics.get("consult")),
                        consult_effective=_convert_numeric(metrics.get("consult_effective")),
                        shopping=_convert_numeric(metrics.get("shopping")),
                        customer_effective=_convert_numeric(metrics.get("customer_effective")),
                        attribution_game_in_app_ltv_1day=_convert_numeric(metrics.get("attribution_game_in_app_ltv_1day")),
                        attribution_game_in_app_roi_1day=_convert_numeric(metrics.get("attribution_game_in_app_roi_1day")),
                        loan_completion=_convert_numeric(metrics.get("loan_completion")),
                        loan_completion_cost=_convert_numeric(metrics.get("loan_completion_cost")),
                        loan_completion_rate=_convert_numeric(metrics.get("loan_completion_rate")),
                        loan_credit=_convert_numeric(metrics.get("loan_credit")),
                        loan_credit_cost=_convert_numeric(metrics.get("loan_credit_cost")),
                        loan_credit_rate=_convert_numeric(metrics.get("loan_credit_rate")),
                        in_app_order_gmv=_convert_numeric(metrics.get("in_app_order_gmv")),
                        in_app_order_roi=_convert_numeric(metrics.get("in_app_order_roi")),
                        in_app_pay_gmv=_convert_numeric(metrics.get("in_app_pay_gmv")),
                        in_app_pay_roi=_convert_numeric(metrics.get("in_app_pay_roi")),
                        total_play=_convert_numeric(metrics.get("total_play")),
                        valid_play=_convert_numeric(metrics.get("valid_play")),
                        valid_play_cost=_convert_numeric(metrics.get("valid_play_cost")),
                        valid_play_rate=_convert_numeric(metrics.get("valid_play_rate")),
                        valid_play_of_mille=_convert_numeric(metrics.get("valid_play_of_mille")),
                        valid_play_cost_of_mille=_convert_numeric(metrics.get("valid_play_cost_of_mille")),
                        average_play_time_per_play=_convert_numeric(metrics.get("average_play_time_per_play")),
                        play_over_rate=_convert_numeric(metrics.get("play_over_rate")),
                        dy_like=_convert_numeric(metrics.get("dy_like")),
                        dy_comment=_convert_numeric(metrics.get("dy_comment")),
                        dy_share=_convert_numeric(metrics.get("dy_share")),
                        report_cnt=_convert_numeric(metrics.get("report_cnt")),
                    )
                    db.add(cost)
                    total_saved += 1

            await db.commit()

            return {
                "success": True,
                "message": f"成功更新 {total_saved} 条记录，共请求 {total_requests} 批次，系统素材总数 {len(system_material_ids)}",
                "count": total_saved
            }

        except Exception as e:
            await db.rollback()
            return {
                "success": False,
                "error": str(e)
            }

#异步获取所有要更新的广告主，素材消耗
async def sync_all_advertisers_consumption(date: str = None):
    """Sync consumption data for all advertisers."""
    if date is None:
        date = (datetime.now(timezone.utc) - timedelta(days=1)).strftime("%Y-%m-%d")
    
    async with async_session() as db:
        result = await db.execute(
            select(UserOAuth.id, UserOAuthAccount.advertiser_id)
            .join(UserOAuthAccount, UserOAuth.id == UserOAuthAccount.oauth_id)
            .join(ResourcesMaterial, UserOAuthAccount.oauth_id == ResourcesMaterial.oauth_id)
            .where(
                UserOAuth.deleted_at.is_(None),
                UserOAuthAccount.deleted_at.is_(None),
                ResourcesMaterial.deleted_at.is_(None),
                ResourcesMaterial.material_id.isnot(None),
                ResourcesMaterial.material_id != "",
            )
            .distinct()
        )
        oauth_advertiser_pairs = result.all()
        
        for oauth_id, advertiser_id in oauth_advertiser_pairs:
            await material_consumption_queue.enqueue({
                "oauth_id": oauth_id,
                "advertiser_id": advertiser_id,
                "date": date,
            })
            logger.info(f"Enqueued consumption sync for oauth_id={oauth_id}, advertiser_id={advertiser_id}, date={date}")
    
    return {"message": f"已将 {len(oauth_advertiser_pairs)} 个广告主的消耗更新任务加入队列"}


material_consumption_queue = MaterialConsumptionQueue()