ASJServer/Controller/CampaignController/AppCampaignController.py

import time
import pytz
from datetime import datetime
from datetime import time as Time

from Model.models import AppAdvertiseCampaign, DeviceTypeModel, CountryModel, Device_User, OpenScreenCampaign
from Ansjer.config import SERVER_TYPE

from django.views import View
from django.db.models import Prefetch

from Object.ResponseObject import ResponseObject
from Object.TokenObject import TokenObject


class AppCampaignView(View):
    def get(self, request, *args, **kwargs):
        request.encoding = 'utf-8'
        operation = kwargs.get('operation')
        request_dict = request.GET
        return self.validation(request, request_dict, operation)

    def post(self, request, *args, **kwargs):
        request.encoding = 'utf-8'
        operation = kwargs.get('operation')
        request_dict = request.POST
        return self.validation(request, request_dict, operation)

    def validation(self, request, request_dict, operation):
        language = request_dict.get('language', 'en')
        response = ResponseObject(language)
        tko = TokenObject(request.META.get('HTTP_AUTHORIZATION'))
        response.lang = tko.lang
        if tko.code != 0:
            return response.json(tko.code)
        userID = tko.userID
        if operation == 'appGetCampaigns':
            return self.app_get_campaigns(userID, request_dict, response)
        elif operation == 'recordUserBehavior':
            return self.record_user_behavior(userID, request_dict)
        else:
            return response.json(414)

    def app_get_campaigns(self, userID, request_dict, response):
        tz = request_dict.get('tz', '+0:00')
        app_type = request_dict.get('appType', 0)
        if SERVER_TYPE == 'Ansjer.cn_config.formal_settings' or SERVER_TYPE == 'Ansjer.cn_config.test_settings':
            s3_url = "https://ansjerfilemanager.s3.cn-northwest-1.amazonaws.com.cn/app/campaign/"
        else:
            s3_url = "https://ansjerfilemanager.s3.amazonaws.com/app/campaign/"

        # 当日时间戳区间获取
        timezone = self.get_timezone_offset(tz)
        current_date = datetime.now().date()
        start_of_day = datetime.combine(current_date, Time.min)
        end_of_day = datetime.combine(current_date, Time.max)
        start_timestamp = int(start_of_day.timestamp())
        end_timestamp = int(end_of_day.timestamp())

        open_screen_campaign_qs = OpenScreenCampaign.objects.filter(
            user_id=userID,
            create_time__gt=start_timestamp,
            create_time__lt=end_timestamp
        )
        if not open_screen_campaign_qs.exists():
            OpenScreenCampaign.objects.create(user_id=userID,
                                              update_time=int(time.time()),
                                              create_time=int(time.time()))
        try:
            country_prefetch = Prefetch('country', queryset=CountryModel.objects.only('id'), to_attr='country_list')
            device_type_prefetch = Prefetch('device_type', queryset=DeviceTypeModel.objects.only('type'),
                                            to_attr='device_type_list')
            app_advertise_campaign_qs = AppAdvertiseCampaign.objects.prefetch_related(country_prefetch,
                                                                                      device_type_prefetch)

            device_info_qs = Device_User.objects.filter(userID=userID).values("region_country").first()
            country_id = device_info_qs.get('region_country')
            if country_id != 0:
                app_advertise_campaign_qs = app_advertise_campaign_qs.filter(country__id=country_id, status=1)
            else:
                app_advertise_campaign_qs = app_advertise_campaign_qs.filter(status=1, unknown_country=1)
            if app_type is not None:
                app_advertise_campaign_qs = app_advertise_campaign_qs.filter(app_bundle_type=app_type)
            # 返回 广告名称、广告类型、开始时间、结束时间、广告图片、活动链接
            campaigns_list = []

            for campaign in app_advertise_campaign_qs:
                # campaign_end_date = (datetime.utcfromtimestamp(campaign.campaign_end_date) # 日期版本
                #                      .replace(tzinfo=pytz.utc).astimezone(timezone).strftime('%Y-%m-%d'))

                # 日期范围
                campaigns_start_date = datetime.fromtimestamp(campaign.campaign_start_date, pytz.utc).astimezone(
                    timezone).replace(hour=0, minute=0, second=0, microsecond=0).timestamp()  # 时间戳版本
                campaign_end_date = datetime.fromtimestamp(campaign.campaign_end_date, pytz.utc).astimezone(
                    timezone).replace(hour=0, minute=0, second=0, microsecond=0).timestamp()  # 时间戳版本

                # 时间范围
                campaign_start_firstday = sum(int(x) * 60 ** i for i, x in enumerate(reversed(
                    datetime.utcfromtimestamp(campaign.campaign_start_date + campaign.campaign_show_stime).replace(
                        tzinfo=pytz.utc).astimezone(timezone).strftime('%H:%M').split(':')))) * 60  # 秒数版本
                campaign_end_firstday = sum(int(x) * 60 ** i for i, x in enumerate(reversed(
                    datetime.utcfromtimestamp(campaign.campaign_start_date + campaign.campaign_show_etime).replace(
                        tzinfo=pytz.utc).astimezone(timezone).strftime('%H:%M').split(':')))) * 60  # 秒数版本

                # campaign_end_firstday = (
                #     datetime.utcfromtimestamp(campaign.campaign_start_date + campaign.campaign_show_etime)
                #     .replace(tzinfo=pytz.utc).astimezone(timezone).strftime('%H:%M')) # 时间版本
                # 轮播广告
                banner_campaign = campaign.banner_campaign
                for banner_cam in banner_campaign:
                    banner_cam["image"] = s3_url + banner_cam["image"]

                campaigns_list.append({
                    'campaign_id': campaign.id,
                    'image_url': s3_url + campaign.image_url,
                    'banner_campaign': banner_campaign,
                    'campaign_url': campaign.campaign_url,
                    'campaign_name': campaign.campaign_name,
                    'campaign_type': campaign.campaign_type,
                    'start_date': campaigns_start_date,
                    'end_date': campaign_end_date,
                    'start_time': campaign_start_firstday,
                    'end_time': campaign_end_firstday,
                    'device_types': [device.type for device in campaign.device_type_list],
                })
            return response.json(0, {
                'interval_time': 86400,
                'campaigns': campaigns_list
            })
        except Exception as e:
            print(e)
            return response.json(500, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e)))

    def record_user_behavior(self, userID, request_dict, response):
        """
        记录用户行为
        @param request_dict: 请求参数
        @param response: 响应对象
        @return: 响应对象
        """
        status = request_dict.get('status', None)  # 1.未跳过 2.已跳过 3.点击广告
        campaign_id = request_dict.get('campaign_id', None)
        user_id = userID

        if not all([status, campaign_id]):
            return response.json(444)

        # 当日时间戳区间获取
        current_date = datetime.now().date()
        start_of_day = datetime.combine(current_date, Time.min)
        end_of_day = datetime.combine(current_date, Time.max)
        start_timestamp = int(start_of_day.timestamp())
        end_timestamp = int(end_of_day.timestamp())

        try:
            # 筛选符合条件的记录
            open_screen_campaigns = OpenScreenCampaign.objects.filter(
                user_id=user_id,
                create_time__gt=start_timestamp,
                create_time__lt=end_timestamp,
                status=0,
            )

            # 检查是否存在记录
            if not open_screen_campaigns.exists():
                # 如果不存在，则创建新记录
                OpenScreenCampaign.objects.create(
                    user_id=user_id,
                    status=status,
                    campaign_id_id=campaign_id,
                    create_time=int(time.time()),
                    update_time=int(time.time())
                )

            else:
                # 如果存在，则更新最新的记录
                latest_campaign = open_screen_campaigns.latest("create_time")
                latest_campaign.status = status
                latest_campaign.campaign_id_id = campaign_id
                latest_campaign.update_time = int(time.time())
                latest_campaign.save()

            return response.json(0)

        except Exception as e:
            print(e)
            return response.json(500, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e)))

    def get_timezone_offset(self, tz):
        """
        将 "+8:00" 格式的时区字符串转换为包含分钟偏移的时区。
        """
        sign = tz[0]  # " " 或 "-"
        hours, minutes = map(int, tz[1:].split('.'))  # 分离小时和分钟

        # 计算总分钟数
        total_minutes = hours * 60 + minutes
        if sign == '-':
            total_minutes = -total_minutes

        # 使用 pytz.FixedOffset 创建时区
        return pytz.FixedOffset(total_minutes)