import os import uuid import pytz import requests import datetime from multiprocessing.spawn import import_main_path from django.views import View from django.conf import settings from django.db import connection from django.utils import timezone from django.contrib import messages from django.shortcuts import render, redirect from django.core.paginator import Paginator, EmptyPage, PageNotAnInteger from django.contrib.auth.mixins import LoginRequiredMixin from django.utils.translation import gettext_lazy as _ from typing import Dict, Any from django.contrib.auth.decorators import login_required from django.db.models import F, ExpressionWrapper, FloatField, Q, Count from django.db.models.functions import Substr from apps.common.utils import get_client_ip from apps.channels.models import Channel from apps.activities.models import Activity from apps.campaigns.models import Campaigns, Adspots from apps.agencies.models import Agency, Brand, Advertiser, BrandCategory, BrandStatus,AgencyType class EmptyAgency: """Empty agency object for template use in create mode.""" def __init__(self): self.pk = None self.id = None self.name = "" self.legal_name = "" self.description = "" self.email = "" self.phone = "" self.website = "" self.founded_year = "" self.agency_size = "" self.employee_count = "" self.street_address = "" self.city = "" self.state = "" self.postal_code = "" self.country = "" self.time_zone = "UTC" self.language = "en" self.logo = None self.agency_type_id = None from apps.core.utils import handle_uploaded_file # ===================================================================== # Agency views # ===================================================================== @login_required def view_agencies(request): # Get the user from the request user = request.user # Get all agencies agencies = Agency.objects.filter( owner=request.user, is_deleted=0 ) # Log user activity Activity.log_activity( user=user, action="view_agencies", description=f"User {user.email if user else 'System'} performed view_agencies", request=request, ip_address=get_client_ip(request), user_agent=request.META.get("HTTP_USER_AGENT", ""), metadata={ } ) return render(request , 'agencies/agency/list.html',{'agency':agencies}) @login_required def add_agency(request): # Get the user from the request user = request.user # check request method if request.method == 'POST': # Get Agency Info name_agency = request.POST.get('name_agency') description_agency = request.POST.get('description_agency') # Check if the channel name is empty if not name_agency: messages.error(request, 'Agency name is required.') return redirect("agencies:add_agency") # Create and save the new agency agency = Agency.objects.create( owner=user, name=name_agency, description=description_agency, datetime=timezone.now() ) # Log user activity Activity.log_activity( user=user, action="add_agency", description=f"User {user.email if user else 'System'} performed add_agency", request=request, ip_address=get_client_ip(request), user_agent=request.META.get("HTTP_USER_AGENT", ""), metadata={ "agency_id": str(agency.id), "agency_name": name_agency, "agency_desc": description_agency, } ) # Redirect to the agency list page return redirect("agencies:view_agencies") return render(request, "agencies/agency/add.html") @login_required def edit_agency(request, id_agency=None): # Get the user from the request user = request.user # check request method if request.method == 'POST': # check if the agency id is not None if id_agency is not None: agency = Agency.objects.get(pk=id_agency) # check if the agency owner is the user if agency.owner != user: messages.error(request, 'You are not the owner of this agency.') return redirect("agencies:edit_agency") data = {'agency':agency} # Log user activity Activity.log_activity( user=user, action="edit_agency", description=f"User {user.email if user else 'System'} performed view_agency with id: {id_agency}", request=request, ip_address=get_client_ip(request), user_agent=request.META.get("HTTP_USER_AGENT", ""), metadata={ "agency_id": str(agency.id), "agency_name": agency.name, "agency_desc": agency.description, } ) return render(request , 'agencies/agency/edit.html',{'data':data}) # Redirect to the agency list page return redirect("agencies:view_agencies") @login_required def update_agency(request, id_agency=None): user = request.user # check request method if request.method == 'POST': # check if the agency id is not None if id_agency is not None: # Get Agency Info name = request.POST.get('name_agency') description = request.POST.get('description_agency') # Check if the agency name is empty if not name : messages.error(request, 'Agency name is required.') return redirect("agencies:edit_agency") # Get Agency Info agency = Agency.objects.get(pk=id_agency) # Update Agency Info agency.name = name if name else agency.name agency.description = description if description else agency.description # Save Agency Info agency.save() # Log user activity Activity.log_activity( user=user, action="edit_agency", description=f"User {user.email if user else 'System'} performed edit_agency with id: {id_agency}", request=request, ip_address=get_client_ip(request), user_agent=request.META.get("HTTP_USER_AGENT", ""), metadata={ "agency_id": str(agency.id), "agency_name": agency.name, "agency_desc": agency.description, } ) # Redirect to the agency list page return redirect("agencies:view_agencies") @login_required def delete_agency(request,id_agency=None): # Get the user from the request user = request.user # check request method if request.method == 'POST': # check if the agency id is not None if id_agency is not None: # Get Agency Info agency = Agency.objects.get(pk=id_agency) # Update Agency Info agency.datetime = datetime.datetime.now() agency.is_deleted = 1 # Save Agency Info agency.save() # Log user activity Activity.log_activity( user=user, action="delete_agency", description=f"User {user.email if user else 'System'} performed delete_agency with id: {id_agency}", request=request, ip_address=get_client_ip(request), user_agent=request.META.get("HTTP_USER_AGENT", ""), metadata={ "agency_id": str(agency.id), "agency_name": agency.name, "agency_desc": agency.description, } ) # Redirect to the agency list page return redirect("agencies:view_agencies") @login_required def load_agency(request): # Get all agencies return render( request, "agencies/components/dropdown_agency_campaign.html", {"agencies": Agency.objects.filter(owner=request.user, is_deleted=0)} ) @login_required def send_agency(request): # Get the user from the request user = request.user name_agency = request.POST.get('name_agency') description_agency = request.POST.get('description_agency') agency = Agency( owner=user, name=name_agency, description=description_agency, datetime=timezone.now(), ) # Save the agency agency.save() return render(request,"agencies/components/dropdown_agency_campaign.html") @login_required def active_agency(request): agences = Agency.objects.filter( owner=request.user, is_deleted=0 ).values_list('id_agency','name','datetime') result = [] for agence in agences: r={} r['name'] = agence[1] brands = Brand.objects.filter(id_agency=agence[0]) campaigns = Campaigns.objects.filter(id_brand__in=brands,pacing=True) if len(campaigns)!=0: r['date'] = agence[2] r['campaigns'] = len(campaigns) result.append(r) return result # ===================================================================== # Advertisers views # ===================================================================== @login_required def views_advertiser(request, id_brand=None): # Get the user from the request user = request.user # if id_brand!="" and id_brand!=None: brand = Brand.objects.get(pk=id_brand) channels = Adspots.objects.filter(brand=brand).values_list('channel').distinct() channels = [ i[0] for i in list(channels)] channels = Channel.objects.filter(pk__in = channels) adspots = Adspots.objects.filter(brand=brand).order_by('channel') # Log user activity Activity.log_activity( user=user, action="view_advertiser", description=f"User {user.email if user else 'System'} performed view_advertiser with id {id_brand}", request=request, ip_address=get_client_ip(request), user_agent=request.META.get("HTTP_USER_AGENT", ""), metadata={ } ) return render( request, 'agencies/advertiser/views_adspots.html', {'data':{ 'adspots':adspots, 'brand': brand , 'channels' : list(channels) }} ) else : brands_tuples = Advertiser.objects.filter(id_user=user).values_list("brand") brands_id = [brand_tuple[0] for brand_tuple in brands_tuples] brands = Brand.objects.filter(pk__in=brands_id).order_by('-created_at') brand_dic = [] for brand in brands: # total_adspots = 0 # try: adspots = Adspots.objects.filter(brand=brand) total_adspots = len(adspots) except Adspots.DoesNotExist: adspot = None # brand_data = { "id_brand" : str(brand.id), "name_brand" : brand.name, "total_adspots" : total_adspots, "logo" : brand.logo_url } brand_dic.append(brand_data) # Log user activity Activity.log_activity( user=user, action="view_advertisers", description=f"User {user.email if user else 'System'} performed view_advertisers", request=request, ip_address=get_client_ip(request), user_agent=request.META.get("HTTP_USER_AGENT", ""), metadata={ } ) return render( request, "agencies/advertiser/views.html", {"brands":brand_dic } ) @login_required def add_advertiser(request): """ Handle brand/advertiser creation with proper validation and error handling """ user = request.user agencies = Agency.objects.filter(owner=request.user,is_deleted=0) if request.method == 'POST': try: # Extract form data with proper field names matching the HTML form name = request.POST.get('name', '').strip() category = request.POST.get('category', '').strip() description = request.POST.get('description', '').strip() status = request.POST.get('status', 'active').strip() is_featured = request.POST.get('is_featured') == 'on' # Checkbox handling id_agency = request.POST.get('agency') # Validation errors = [] # Required field validation if not name: errors.append("Brand name is required.") elif len(name) > 255: errors.append("Brand name cannot exceed 255 characters.") if not category: errors.append("Category is required.") elif category not in dict(BrandCategory.choices): errors.append("Invalid category selected.") if status not in dict(BrandStatus.choices): errors.append("Invalid status selected.") if description and len(description) > 1000: errors.append("Description cannot exceed 1000 characters.") # Handle logo upload logo_file = None if 'logo' in request.FILES: logo_file = request.FILES['logo'] # Validate file size (5MB limit) max_size = 5 * 1024 * 1024 # 5MB if logo_file.size > max_size: errors.append("Logo file size cannot exceed 5MB.") # Validate file format allowed_extensions = ['jpg', 'jpeg', 'png', 'svg', 'webp'] file_extension = logo_file.name.split('.')[-1].lower() if file_extension not in allowed_extensions: errors.append("Logo must be in JPG, JPEG, PNG, SVG, or WebP format.") # If there are validation errors, return to form with errors if errors: for error in errors: messages.error(request, error) return redirect("agencies:add_advertiser") # Get the user's agency (assuming there's a relationship) try: # agency = Agency.objects.filter(pk=id_agency) # if agency.count() <= 0: # Adjust this based on your user-agency relationship agency = Agency.objects.filter( owner=request.user, is_deleted=0 ) except AttributeError: messages.error(request, "User is not associated with any agency.") return redirect("agencies:views_advertiser") print(agency) print(agency.count()) agency = agency.first() # Check for duplicate brand name within the same agency brand_exist = Brand.objects.filter(name=name, agency=agency).exists() if not brand_exist: # Create the Brand instance brand = Brand( name=name, agency=agency, description=description, category=category, status=status, is_featured=is_featured ) # Handle logo file upload if provided if logo_file: # Generate unique filename to prevent conflicts file_extension = logo_file.name.split('.')[-1].lower() unique_filename = f"{uuid.uuid4().hex}_{name[:50]}.{file_extension}" # Use Django's file handling for better security and management brand.logo.save(unique_filename, logo_file, save=False) # Save the brand brand.save() # Log activity (if Activity model exists) try: # Log user activity Activity.log_activity( user=user, action="create_brand", description=f"User {user.email if user else 'System'} performed create_brand", request=request, ip_address=get_client_ip(request), user_agent=request.META.get("HTTP_USER_AGENT", ""), metadata={ "brand": str(brand.id) } ) except ImportError: pass # Activity model doesn't exist else: brand = Brand.objects.get(name=name, agency=agency) # Create Advertiser relationship (if Advertisers model exists) try: advertiser_user = Advertisers( id_user=user, brand=brand, status=status, ) advertiser_user.save() except ImportError: pass # Advertisers model doesn't exist # Log activity (if Activity model exists) try: # Log user activity Activity.log_activity( user=user, action="create_advertiser", description=f"User {user.email if user else 'System'} performed create_advertiser", request=request, ip_address=get_client_ip(request), user_agent=request.META.get("HTTP_USER_AGENT", ""), metadata={ "create_advertiser": str(advertiser_user.id) } ) except ImportError: pass # Activity model doesn't exist # Success message and redirect messages.success(request, f'Brand "{brand.name}" has been created successfully!') return redirect("agencies:views_advertiser") except Exception as e: # Handle unexpected errors messages.error(request, f"An error occurred while creating the brand: {str(e)}") return redirect("agencies:add_advertiser") else: # GET request - render the form context = { "agencies": agencies, "brand_categories": BrandCategory.choices, "brand_statuses": BrandStatus.choices, } return render(request, "agencies/advertiser/add.html", context) @login_required def edit_advertiser(request,id_brand=None): print(request.method) if request.method == 'POST': print(id_brand) brand = Brand.objects.get(pk=id_brand) return render(request , 'core/edit_advertiser_id.html', {'brand' : brand}) else: # id_brand = Advertisers.objects.filter(id_user=request.session['id_user']).select_related('id_brand').all() # brands = Brand.objects.filter(pk__in=brands_id) brands_id = Advertisers.objects.filter(id_user=request.user).values_list('id_brand') brands_id = [x[0] for x in brands_id] brands = Brand.objects.filter(pk__in=brands_id).order_by('-id_brand') return render(request,'agencies/advertiser/edit.html',{ 'brands' : brands}) @login_required def update_advertiser(request , id_brand=None): brand = Brand.objects.get(pk=id_brand) advertiser_name = request.POST.get('advertiser_name') category = request.POST.get('category') advertiser_desc = request.POST.get('advertiser_desc') active_0_1 = request.POST.get('active_0_1') now = datetime.datetime.now() try : path = "Advertisers_imgs" now_infile = str(now).replace(" ", "_") now_infile = now_infile.replace(":", "-") filename = advertiser_name+"__"+now_infile+".png" handle_uploaded_file(request.FILES['advertiser_logo'], 'static/'+path, filename) brand.logo = 'static/'+path+'/'+filename except : pass if advertiser_name != '' : brand.brand_name = advertiser_name if category != '': brand.category = category if advertiser_desc != '': brand.description = advertiser_desc if active_0_1 != '': brand.status = active_0_1 activite = 'Edit Advertirer' desc = 'Admin edit advertiser id: ' + str(brand.id_brand) activity = Activity(activity=activite , date=now ,description=desc ) activity.save() brand.save() brands = Advertisers.objects.filter(id_user=request.session['id_user']).filter(status="1").select_related('id_brand').all() return redirect("agencies:views_advertiser") @login_required def disable_advertiser(request , id_brand=None): if request.method == 'POST': now = datetime.datetime.now() activite = 'Disable Advertirer' desc = 'Admin disable advertiser id: ' + str(id_brand) activity = Activity(activity=activite , date=now ,description=desc ) activity.save() brand = Brand.objects.get(pk=id_brand) status = (1+ int(brand.status) ) % 2 brand.status = str(status) brand.save() # brands_id = Advertisers.objects.filter(id_user=request.session['id_user']).select_related('id_brand').all() # brands = Brand.objects.filter(pk__in=brands_id).order_by('-id_brand') brands_id = Advertisers.objects.filter(id_user=request.session['id_user']).values_list('id_brand') brands_id = [x[0] for x in brands_id] brands = Brand.objects.filter(pk__in=brands_id).order_by('-id_brand') return render(request,'core/advertiser/disable.html',{ 'brands' : brands}) @login_required def load_advertisers(request): agency_id = request.GET.get('agency') advertisers = Brand.objects.filter(id_agency=agency_id) return render(request,'core/dropdown_advertisers_campaign.html',{'brands': advertisers}) @login_required def send_advertisers(request): advertiser_name = request.POST.get('name') print(advertiser_name) category = request.POST.get('category') print(category) advertiser_desc = request.POST.get('description') active_0_1 = request.POST.get('status') now = datetime.datetime.now() path = "static/Advertisers_imgs" now_infile = str(now).replace(" ", "_") now_infile = now_infile.replace(":", "-") filename = advertiser_name+"__"+now_infile+".png" # handle_uploaded_file(request.FILES['logo'], path, filename) agency = request.POST.get('agency') agency = Agency.objects.get(id_agency=agency) Brand = Brand( id_agency=agency, brand_name=advertiser_name , description=advertiser_desc, category=category, status=active_0_1, logo=path+'/'+filename ) Brand.save() Advertiser_user = Advertisers( id_user_id=request.session['id_user'], id_brand_id=Brand.id_brand, status=active_0_1, datetime=now ) Advertiser_user.save() return render(request,'core/dropdown_advertisers_campaign.html') @login_required def most_advertisers(request) : brands_id = Advertisers.objects.filter(id_user=request.session['id_user']).values_list('id_brand') brands_id = [x[0] for x in brands_id] brands = Brand.objects.filter(pk__in=brands_id).order_by('-id_brand') totals = [] dic = [] result = [] for i in brands: adspots = Adspots.objects.filter(brand=i) toutal = len(adspots) totals.append(toutal) brand = {'id_brand' : i.id_brand, 'name_brand' : i.brand_name, 'total_adspots' : toutal, 'logo' : i.logo } print(brand) dic.append(brand) print(totals) for i in range(0,4): max_value = max(totals) max_index = totals.index(max_value) a=totals.pop(max_index) result.append(dic[max_index]) print(result) return result @login_required def report_advertiser_backup(request ,id_brand=""): from django.db.models import F, ExpressionWrapper, FloatField from django.db.models.functions import Substr from datetime import timedelta # def my_custom_sql(query, params): # from django.db import connection # with connection.cursor() as cursor: # cursor.execute(query, params) # columns = [col[0] for col in cursor.description] # return [ # dict(zip(columns, row)) # for row in cursor.fetchall() # ] def dictfetchall(cursor): "Return all rows from a cursor as a dict" columns = [col[0] for col in cursor.description] return [ dict(zip(columns, row)) for row in cursor.fetchall() ] def my_custom_sql(query, params): try: with connection.cursor() as cursor: cursor.execute(query, params) rows = cursor.fetchall() return rows except Exception as e: print(f"An error occurred: {e}") return [] if request.method =='POST': from datetime import datetime import pytz # Get the current date and time current_datetime = datetime.now() # Format the date in 'YYYY-MM-DD' format report_date = current_datetime.strftime('%Y-%m-%d') paris_tz = pytz.timezone('Europe/Paris') paris_time = datetime.now(paris_tz) utc_offset = paris_time.utcoffset().total_seconds() / 3600 print(f"Current UTC offset for France (Paris) is GMT+{int(utc_offset)}.") start_day = request.POST.get('start_day').replace("-", "") end_day = request.POST.get('end_day').replace("-", "") selected_brand = request.POST.get('selected_brand') # if utc_offset == 1: # # # ************************************* # # * for when France timezone is GMT+1 * # # ************************************* # # query = """ # SELECT networkname, spotId, airTime, airStatusCode, purcent AS sfr_percentage, purcent*1.25*4500000/17 AS total_volume, Epg.emission_name # FROM Verifs # LEFT JOIN SFR_analytics ON SUBSTRING(Verifs.airTime, 12, 5) = SUBSTRING(SFR_analytics.`minute`, 1, 5) # AND SUBSTRING(Verifs.airTime, 1, 10) = SUBSTRING(SFR_analytics.`day`, 1, 10) # AND SFR_analytics.sfr_channel_name = '2M Maroc' # LEFT JOIN Epg ON (Verifs.airTime < Epg.End_time AND Verifs.airTime > Epg.Start_time) # WHERE Verifs.spotId LIKE %s AND Verifs.airStatusCode = '0001' AND Verifs.broadcastDate > %s AND Verifs.broadcastDate < %s # GROUP BY networkname, spotId, airTime, airStatusCode, purcent, Epg.emission_name ORDER BY airTime # """ # if utc_offset == 2: if utc_offset == 1: # ************************************* # * for when France timezone is GMT+2 * # ************************************* query_old = """ SELECT networkname, spotId, DATE_SUB(Verifs.airTime, INTERVAL 1 HOUR) as airTime, airStatusCode, purcent AS sfr_percentage, purcent*1.25*4500000/17 AS total_volume, Epg.emission_name FROM Verifs LEFT JOIN SFR_analytics ON SUBSTRING(Verifs.airTime, 12, 5) = SUBSTRING(SFR_analytics.`minute`, 1, 5) AND SUBSTRING(Verifs.airTime, 1, 10) = SUBSTRING(SFR_analytics.`day`, 1, 10) AND SFR_analytics.sfr_channel_name = '2M Maroc' LEFT JOIN Epg ON (Verifs.airTime < DATE_ADD(Epg.End_time, INTERVAL 1 HOUR) AND Verifs.airTime > DATE_ADD(Epg.Start_time, INTERVAL 1 HOUR)) WHERE Verifs.spotId LIKE %s AND Verifs.airStatusCode = '0001' AND Verifs.broadcastDate > %s AND Verifs.broadcastDate < %s GROUP BY networkname, spotId, airTime, airStatusCode, purcent, Epg.emission_name ORDER BY airTime """ query_ol3 = """ SELECT networkname, spotId, DATE_SUB(Verifs.airTime, INTERVAL 1 HOUR) as airTime, airStatusCode, purcent AS sfr_percentage, CAST(purcent * 1.25 * 4500000 / 17 AS UNSIGNED) AS total_volume, Epg.emission_name FROM Verifs LEFT JOIN SFR_analytics ON SUBSTRING(Verifs.airTime, 12, 5) = SUBSTRING(SFR_analytics.`minute`, 1, 5) AND SUBSTRING(Verifs.airTime, 1, 10) = SUBSTRING(SFR_analytics.`day`, 1, 10) AND SFR_analytics.sfr_channel_name = '2M Maroc' LEFT JOIN Epg ON (Verifs.airTime < DATE_ADD(Epg.End_time, INTERVAL 1 HOUR) AND Verifs.airTime > DATE_ADD(Epg.Start_time, INTERVAL 1 HOUR)) WHERE Verifs.spotId LIKE %s AND Verifs.airStatusCode = '0001' AND Verifs.broadcastDate > %s AND Verifs.broadcastDate < %s GROUP BY networkname, spotId, airTime, airStatusCode, purcent, Epg.emission_name ORDER BY airTime """ query_old4 = """ SELECT networkname, spotId, DATE_SUB(Verifs.airTime, INTERVAL 1 HOUR) as airTime, airStatusCode, purcent AS sfr_percentage, CAST(purcent * 1.25 * 4500000 / 17 AS UNSIGNED) AS total_volume, Epg.emission_name FROM Verifs LEFT JOIN SFR_analytics ON SUBSTRING(Verifs.airTime, 12, 5) = SUBSTRING(SFR_analytics.`minute`, 1, 5) AND SUBSTRING(Verifs.airTime, 1, 10) = SUBSTRING(SFR_analytics.`day`, 1, 10) AND SFR_analytics.sfr_channel_name = '2M Maroc' Inner JOIN Epg ON (Verifs.airTime < DATE_ADD(Epg.End_time, INTERVAL 2 HOUR) AND Verifs.airTime > DATE_ADD(Epg.Start_time, INTERVAL 2 HOUR)) WHERE Verifs.spotId LIKE %s AND Verifs.airStatusCode = '0001' AND Verifs.broadcastDate > %s AND Verifs.broadcastDate < %s GROUP BY networkname, spotId, airTime, airStatusCode, purcent, Epg.emission_name ORDER BY airTime """ query = """ SELECT networkname, spotId, DATE_SUB(Verifs.airTime, INTERVAL 1 HOUR) as airTime, airStatusCode, purcent AS sfr_percentage, CAST(MAX(purcent) * 1.25 * 4500000 / 17 AS UNSIGNED) AS total_volume, Epg.emission_name FROM Verifs LEFT JOIN ( SELECT sfr_channel_name, `minute`, `day`, MAX(purcent) as purcent FROM SFR_analytics WHERE sfr_channel_name = '2M Maroc' GROUP BY sfr_channel_name, `minute`, `day` ) AS SFR_analytics ON SUBSTRING(Verifs.airTime, 12, 5) = SUBSTRING(SFR_analytics.`minute`, 1, 5) AND SUBSTRING(Verifs.airTime, 1, 10) = SUBSTRING(SFR_analytics.`day`, 1, 10) Inner JOIN Epg ON (Verifs.airTime < DATE_ADD(Epg.End_time, INTERVAL 2 HOUR) AND Verifs.airTime > DATE_ADD(Epg.Start_time, INTERVAL 2 HOUR)) WHERE Verifs.spotId LIKE %s AND Verifs.airStatusCode = '0001' AND Verifs.broadcastDate > %s AND Verifs.broadcastDate < %s GROUP BY networkname, spotId, airTime, airStatusCode, purcent, Epg.emission_name ORDER BY airTime """ # query = """ # SELECT networkname, spotId, airTime, airStatusCode, purcent AS sfr_percentage, purcent*1.25*4500000/17 AS total_volume, Epg.emission_name # FROM Verifs # LEFT JOIN SFR_analytics ON SUBSTRING(Verifs.airTime, 12, 5) = SUBSTRING(SFR_analytics.`minute`, 1, 5) # AND SUBSTRING(Verifs.airTime, 1, 10) = SUBSTRING(SFR_analytics.`day`, 1, 10) # AND SFR_analytics.sfr_channel_name = '2M Maroc' # LEFT JOIN Epg ON (Epg.End_time > DATE_SUB(Verifs.airTime, INTERVAL 1 HOUR) AND Epg.End_time < DATE_SUB(Verifs.airTime, INTERVAL 1 HOUR)) # WHERE Verifs.spotId LIKE %s AND Verifs.airStatusCode = '0001' AND Verifs.broadcastDate > %s AND Verifs.broadcastDate < %s # GROUP BY networkname, spotId, airTime, airStatusCode, purcent, Epg.emission_name ORDER BY airTime # """ # The % signs are part of the parameter, not the SQL string # params = [f"%{selected_brand}%", '0001%', start_day, end_day] from django.db import connection cursor = connection.cursor() data_tuple=(f"%{selected_brand}%", start_day, end_day) cursor.execute(query,data_tuple) # data = cursor.fetchall() total = 0 data = dictfetchall(cursor) processed_data = [] last_date = None for row in data: # Convert airTime string to datetime object to extract the date part only row_date = datetime.strptime(row['airTime'], '%Y-%m-%d %H:%M:%S').date() if last_date and row_date != last_date: # Insert an empty or placeholder row processed_data.append({'networkname': '--', 'spotId': '--', 'airTime': '--', 'total_volume': '--', 'emission_name': '--'}) processed_data.append(row) last_date = row_date for row in data: if row['total_volume'] == None: print(row) else: total += int(row['total_volume']) return render(request,'core/view_report.html', {'data': processed_data, 'total':total, 'selected_brand':selected_brand, 'report_date':report_date }) # result = Verifs.objects.filter(spotId__contains='kia', airStatuscode__startswith='0001', broadcastDate__gt=start_date)\ # .annotate(sfr_percentage=F('purcent'), total_volume=ExpressionWrapper(F('purcent') * 1.25 * 4500000 / 17, output_field=DateTimeField()))\ # .values('networkname', 'spotId', 'airTime', 'airStatusCode', 'sfr_percentage', 'total_volume', 'epg__emission_name')\ # .prefetch_related('sfranalytics_set', 'epg') # # # Apply the JOIN conditions # result = result.filter(sfranalytics__minute__startswith=Substr(F('airTime'), 12, 5), # sfranalytics__day=Substr(F('airTime'), 1, 10), # sfranalytics__sfr_channel_name='2M Maroc', # epg__start_time__lt=ExpressionWrapper(F('airTime') + timedelta(hours=1), output_field=DateTimeField()), # epg__end_time__gt=ExpressionWrapper(F('airTime') + timedelta(hours=1), output_field=DateTimeField())) # # Retrieve the required fields # result = result.values('networkname', 'spotId', 'airTime', 'airStatusCode', 'sfr_percentage', # 'total_volume', 'epg__emission_name') # # # Apply the ordering # result = result.order_by('airTime') # ptint(result) # return render(request,'core/views_advertiser_adspots.html', {'data':{'adspots':adspots, 'brand' : brand , 'channels' : list(channels)}}) else : brands_id = Advertiser.objects.filter(id_user=request.session['id_user']).values_list('id_brand') brands_id = [x[0] for x in brands_id] brands = Brand.objects.filter(pk__in=brands_id, status=1).order_by('-id_brand') dic = [] for i in brands: adspots = Adspots.objects.filter(brand=i) toutal = len(adspots) brand = {'id_brand' : i.id_brand, 'name_brand' : i.brand_name, 'description' : i.description, 'total_adspots' : toutal, 'logo' : i.logo } print(brand) dic.append(brand) return render(request , 'core/advertiser/report.html' , {'brands':dic } ) @login_required def report_advertiser(request, id_brand=""): def dictfetchall(cursor): "Return all rows from a cursor as a dict" columns = [col[0] for col in cursor.description] return [ dict(zip(columns, row)) for row in cursor.fetchall() ] def my_custom_sql(query, params): try: with connection.cursor() as cursor: cursor.execute(query, params) rows = cursor.fetchall() return rows except Exception as e: print(f"An error occurred: {e}") return [] if request.method == 'POST': current_datetime = datetime.now() report_date = current_datetime.strftime('%Y-%m-%d') paris_tz = pytz.timezone('Europe/Paris') paris_time = datetime.now(paris_tz) utc_offset = paris_time.utcoffset().total_seconds() / 3600 print(f"Current UTC offset for France (Paris) is GMT+{int(utc_offset)}.") start_day = request.POST.get('start_day').replace("-", "") end_day = request.POST.get('end_day').replace("-", "") selected_brand = request.POST.get('selected_brand') if utc_offset == 2: # normal one hardcoded for 2M: query_old = """ SELECT networkname, spotId, DATE_SUB(Verifs.airTime, INTERVAL 1 HOUR) as airTime, airStatusCode, purcent AS sfr_percentage, CAST(MAX(purcent) * 1.25 * 4500000 / 17 AS UNSIGNED) AS total_volume, Epg.emission_name FROM Verifs LEFT JOIN ( SELECT sfr_channel_name, `minute`, `day`, MAX(purcent) as purcent FROM SFR_analytics WHERE sfr_channel_name = '2M Maroc' GROUP BY sfr_channel_name, `minute`, `day` ) AS SFR_analytics ON SUBSTRING(Verifs.airTime, 12, 5) = SUBSTRING(SFR_analytics.`minute`, 1, 5) AND SUBSTRING(Verifs.airTime, 1, 10) = SUBSTRING(SFR_analytics.`day`, 1, 10) Inner JOIN Epg ON (Verifs.airTime < DATE_ADD(Epg.End_time, INTERVAL 2 HOUR) AND Verifs.airTime > DATE_ADD(Epg.Start_time, INTERVAL 2 HOUR)) WHERE Verifs.spotId LIKE %s AND Verifs.airStatusCode = '0001' AND Verifs.broadcastDate > %s AND Verifs.broadcastDate < %s GROUP BY networkname, spotId, airTime, airStatusCode, purcent, Epg.emission_name ORDER BY airTime """ query_11 = """ SELECT networkname, spotId, DATE_SUB(Verifs.airTime, INTERVAL 1 HOUR) as airTime, airStatusCode, purcent AS sfr_percentage, CAST(MAX(purcent) * 1.25 * 4500000 / CASE WHEN DATE(Verifs.airTime) < '2024-03-18' THEN 17 ELSE 14 END AS UNSIGNED) AS total_volume, Epg.emission_name FROM Verifs LEFT JOIN ( SELECT sfr_channel_name, `minute`, `day`, MAX(purcent) as purcent FROM SFR_analytics WHERE sfr_channel_name = '2M Maroc' GROUP BY sfr_channel_name, `minute`, `day` ) AS SFR_analytics ON SUBSTRING(Verifs.airTime, 12, 5) = SUBSTRING(SFR_analytics.`minute`, 1, 5) AND SUBSTRING(Verifs.airTime, 1, 10) = SUBSTRING(SFR_analytics.`day`, 1, 10) INNER JOIN Epg ON (Verifs.airTime < DATE_ADD(Epg.End_time, INTERVAL 2 HOUR) AND Verifs.airTime > DATE_ADD(Epg.Start_time, INTERVAL 2 HOUR) AND Epg.verif_channel_id = Verifs.zonename) WHERE Verifs.spotId LIKE %s AND Verifs.airStatusCode = '0001' AND Verifs.broadcastDate > %s AND Verifs.broadcastDate < %s GROUP BY networkname, spotId, airTime, airStatusCode, purcent, Epg.emission_name ORDER BY airTime """ query11 = """ Select MIN(networkname) AS networkname, MIN(networkname) AS networkname, MIN(spotId) AS spotId, MIN(Epg.emission_name) AS emission_name, SUBSTRING(DATE_SUB(Verifs.airTime, INTERVAL 2 HOUR), 1, 19) as airTime, CAST(MAX(purcent) * 1.25 * 4500000 / CASE WHEN DATE(Verifs.broadcastDate) < '20240317' THEN 17 ELSE 14 END AS UNSIGNED) AS total_volume FROM Verifs INNER JOIN SFR_analytics ON SUBSTRING(Verifs.airTime, 12, 5) = SUBSTRING(SFR_analytics.`minute`, 1, 5) AND SUBSTRING(Verifs.airTime, 1, 10) = SUBSTRING(SFR_analytics.`day`, 1, 10) AND SFR_analytics.sfr_channel_name = '2M Maroc' INNER JOIN Epg ON (Verifs.airTime < DATE_ADD(Epg.End_time, INTERVAL 1 HOUR) AND Verifs.airTime > DATE_ADD(Epg.Start_time, INTERVAL 1 HOUR)) WHERE Verifs.spotId LIKE %s AND Verifs.airStatusCode = '0001' AND Verifs.broadcastDate > %s AND Verifs.broadcastDate < %s GROUP BY SUBSTRING(Verifs.airTime, 1, 19), Verifs.broadcastDate ORDER BY airTime; """ #made for Ramadan 2024 query = """ SELECT MIN(Verifs.networkname) AS networkname, MIN(spotId) AS spotId, CASE WHEN DATE(Verifs.airTime) > '2024-03-30' THEN SUBSTRING(Verifs.airTime, 1, 19) ELSE SUBSTRING(DATE_SUB(Verifs.airTime, INTERVAL 1 HOUR), 1, 19) END AS airTime, MIN(Epg.emission_name) AS emission_name, CAST(MAX(purcent) * 1.25 * 4500000 / CASE WHEN DATE(Verifs.broadcastDate) < '2024-03-17' THEN Channels_zone.market_share ELSE Channels_zone.market_share END AS UNSIGNED) AS total_volume FROM Verifs INNER JOIN Channels_zone ON Channels_zone.id_zone_channel = Verifs.zonename Inner JOIN ( SELECT sfr_channel_name, `minute`, `day`, MAX(purcent) AS purcent FROM SFR_analytics INNER JOIN Channels_zone ON Channels_zone.sfr_name = SFR_analytics.sfr_channel_name WHERE sfr_channel_name = Channels_zone.sfr_name GROUP BY sfr_channel_name, `minute`, `day` ) AS SFR_analytics ON ( SUBSTRING(Verifs.airTime, 12, 5) = SUBSTRING(SFR_analytics.`minute`, 1, 5) OR SUBSTRING(Verifs.airTime, 12, 5) = SUBSTRING(ADDTIME(SFR_analytics.`minute`, '00:01:00'), 1, 5) ) AND SUBSTRING(Verifs.airTime, 1, 10) = SUBSTRING(SFR_analytics.`day`, 1, 10) AND Channels_zone.sfr_name = SFR_analytics.sfr_channel_name INNER JOIN Epg ON ( Epg.verif_channel_id = Verifs.networkname AND Verifs.airTime < DATE_ADD(Epg.End_time, INTERVAL 2 HOUR) AND Verifs.airTime > DATE_ADD(Epg.Start_time, INTERVAL 2 HOUR) ) WHERE Verifs.spotId LIKE %s AND Verifs.airStatusCode = '0001' AND Verifs.broadcastDate > %s AND Verifs.broadcastDate < %s GROUP BY CASE WHEN DATE(Verifs.airTime) > '2024-03-30' THEN SUBSTRING(Verifs.airTime, 1, 19) ELSE SUBSTRING(DATE_SUB(Verifs.airTime, INTERVAL 1 HOUR), 1, 19) END, Verifs.broadcastDate, Channels_zone.market_share, SFR_analytics.sfr_channel_name ORDER BY airTime; """ cursor = connection.cursor() selected_brand = selected_brand.capitalize() data_tuple = (f"%{selected_brand}%", start_day, end_day) cursor.execute(query, data_tuple) data = dictfetchall(cursor) processed_data = [] last_date = None total = 0 for row in data: row_date = datetime.strptime(row['airTime'], '%Y-%m-%d %H:%M:%S').date() if last_date and row_date != last_date: processed_data.append({'networkname': '--', 'spotId': '--', 'airTime': '--', 'total_volume': '--', 'emission_name': '--'}) processed_data.append(row) last_date = row_date if row['total_volume'] is not None: total += int(row['total_volume']) return render(request, 'core/view_report.html', {'data': processed_data, 'total': total, 'selected_brand': selected_brand, 'report_date': report_date}) else: brands_id = Advertisers.objects.filter(id_user=request.session['id_user']).values_list('id_brand', flat=True) brands = Brands.objects.filter(pk__in=brands_id, status=1).order_by('-id_brand') dic = [] for brand in brands: adspots = Adspots.objects.filter(brand=brand) total_adspots = len(adspots) brand_info = { 'id_brand': brand.id_brand, 'name_brand': brand.brand_name, 'description': brand.description, 'total_adspots': total_adspots, 'logo': brand.logo } dic.append(brand_info) return render(request, 'core/report_advertiser.html', {'brands': dic}) # ===================================================================== # Brands views # ===================================================================== @login_required def statsbrands(request): from django.db.models import Count ch = Channel.objects.filter(id_user=1) ads = Campaigns.objects.filter(id_user=request.session['id_user']).values('id_brand').annotate(dcount=Count("id_brand")).order_by('-dcount') label = [] data = [] import random chars = '0123456789ABCDEF' color = ['#'+''.join(random.sample(chars,6)) for i in range(len(ads))] for ad in ads : label.append(Brands.objects.get(id_brand=ad['id_brand']).brand_name) data.append(ad['dcount']) brands={ 'label' : label , 'data' : data , 'color' : color } return render(request ,'core/stats_brands.html', { 'brands' : brands} ) def brand_adspot(): from django.db.models import Count ch = Channel.objects.filter(id_user=1) verifs = Verifs.objects.filter(networkname__in = ch , airStatuscode='0001' ).values_list('spotId') verifs = [verif[0] for verif in verifs ] ads = Adspots.objects.filter(name__in=v, channel__in=ch ).values('brand').annotate(dcount=Count("brand")).order_by('-dcount') import random chars = '0123456789ABCDEF' color = ['#'+''.join(random.sample(chars,6)) for i in range(len(ads))] label = [] data = [] for ad in ads : label.append(Brands.objects.get(id_brand=ad['id_brand']).brand_name) data.append(ad['dcount']) # ===================================================================== # New class-based views for Agencies (modeled after Channels CBVs) # ===================================================================== from django.contrib.auth.mixins import LoginRequiredMixin from django.core.paginator import Paginator, PageNotAnInteger, EmptyPage from django.http import HttpRequest, HttpResponse, JsonResponse, Http404 from django.shortcuts import get_object_or_404 from django.urls import reverse class AgencyListView(LoginRequiredMixin, View): """ Agency List View This view is responsible for rendering the agencies list page of the application. It extends the Django View class and defines the HTTP GET method to handle the request and return the rendered agencies list page template with advanced filtering, searching, pagination, and AJAX support. Methods: get(request): Handles the HTTP GET request and returns the rendered agencies list page template. Template: - 'agencies/agency/list.html': The HTML template used to render the agencies list page. Features: - Advanced filtering by agency type, status, size, and date ranges - Full-text search across agency name, legal name, and description - Pagination with customizable items per page - AJAX support for dynamic updates - Statistics and analytics - Export functionality - Real-time updates """ http_method_names = ["get"] template_name = "agencies/agency/list.html" extra_context = { "title": "Agencies - Focus", "author": "Focus Development Team", "description": "Agency management for Focus advertising solutions. This page provides comprehensive agency management and insights.", "keywords": "Focus, agencies, advertising solutions, agency management, marketing, data visualization", "active_menu": "agencies", } def get_context_data(self, **kwargs): """ Add agencies page context data. Args: **kwargs: Additional keyword arguments Returns: dict: Updated context data """ context = super().get_context_data(**kwargs) context.update({ "title": _("Agencies"), }) return context def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse: """ Handles the HTTP GET request and returns the rendered agencies list page template. Args: request (HttpRequest): The HTTP request object. *args: Variable length argument list. **kwargs: Arbitrary keyword arguments. Returns: HttpResponse: The rendered agencies list page template. Query Parameters: - search: Search term for agency name/legal_name/description - agency_type: Filter by agency type ID - status: Filter by agency status - size: Filter by agency size - date_from: Start date for filtering - date_to: End date for filtering - date_range: Predefined date range - sort_by: Sorting field and direction - page: Page number for pagination - per_page: Number of items per page (default: 20) """ # Check for AJAX requests if request.headers.get('X-Requested-With') == 'XMLHttpRequest': return self._handle_ajax_request(request) # Get filtered queryset queryset = self._get_filtered_queryset(request) # Apply pagination paginator = Paginator(queryset, self._get_items_per_page(request)) page_number = request.GET.get('page', 1) try: page_obj = paginator.get_page(page_number) except (PageNotAnInteger, EmptyPage): page_obj = paginator.get_page(1) # Log user activity for auditing self._log_user_activity(request) # Build context context = self._build_context(request, page_obj, paginator) return render(request, self.template_name, context) def _get_filtered_queryset(self, request: HttpRequest): """ Build filtered queryset based on request parameters. Args: request: Django HTTP request object Returns: QuerySet: Filtered agency queryset """ # Base queryset with optimizations queryset = Agency.objects.all().select_related('agency_type').prefetch_related('team_members') # Scope to owner unless superuser if not request.user.is_superuser: queryset = queryset.filter(owner=request.user) # Apply filters queryset = self._apply_search_filter(queryset, request) queryset = self._apply_agency_type_filter(queryset, request) queryset = self._apply_status_filter(queryset, request) queryset = self._apply_size_filter(queryset, request) queryset = self._apply_date_filters(queryset, request) # Apply sorting queryset = self._apply_sorting(queryset, request) return queryset def _apply_search_filter(self, queryset, request): """ Apply search filter to queryset. Args: queryset: Base queryset request: Django HTTP request object Returns: QuerySet: Filtered queryset """ search_query = request.GET.get('search', '').strip() if search_query: queryset = queryset.filter( Q(name__icontains=search_query) | Q(legal_name__icontains=search_query) | Q(description__icontains=search_query) | Q(email__icontains=search_query) ) return queryset def _apply_agency_type_filter(self, queryset, request): """ Apply agency type filter to queryset. Args: queryset: Base queryset request: Django HTTP request object Returns: QuerySet: Filtered queryset """ agency_type = request.GET.get('agency_type', '').strip() if agency_type and agency_type != 'all': try: agency_type_id = int(agency_type) queryset = queryset.filter(agency_type_id=agency_type_id) except (ValueError, TypeError): pass return queryset def _apply_status_filter(self, queryset, request): """ Apply status filter to queryset. Args: queryset: Base queryset request: Django HTTP request object Returns: QuerySet: Filtered queryset """ status_filter = request.GET.get('status', '').strip() if status_filter and status_filter != 'all': if status_filter == 'active': queryset = queryset.filter(is_active=True, is_deleted=False) elif status_filter == 'inactive': queryset = queryset.filter(is_active=False, is_deleted=False) elif status_filter == 'verified': queryset = queryset.filter(is_verified=True, is_deleted=False) elif status_filter == 'featured': queryset = queryset.filter(is_featured=True, is_deleted=False) elif status_filter == 'deleted': queryset = queryset.filter(is_deleted=True) else: # Default to non-deleted queryset = queryset.filter(is_deleted=False) return queryset def _apply_size_filter(self, queryset, request): """ Apply agency size filter to queryset. Args: queryset: Base queryset request: Django HTTP request object Returns: QuerySet: Filtered queryset """ size_filter = request.GET.get('size', '').strip() if size_filter and size_filter != 'all': queryset = queryset.filter(agency_size=size_filter) return queryset def _apply_date_filters(self, queryset, request): """ Apply date range filters to queryset. Args: queryset: Base queryset request: Django HTTP request object Returns: QuerySet: Filtered queryset """ date_from = request.GET.get('date_from') date_to = request.GET.get('date_to') date_range = request.GET.get('date_range') # Handle predefined date ranges if date_range: today = timezone.now().date() if date_range == 'today': queryset = queryset.filter(created_at__date=today) elif date_range == 'yesterday': yesterday = today - datetime.timedelta(days=1) queryset = queryset.filter(created_at__date=yesterday) elif date_range == 'this_week': week_start = today - datetime.timedelta(days=today.weekday()) queryset = queryset.filter(created_at__date__gte=week_start) elif date_range == 'this_month': month_start = today.replace(day=1) queryset = queryset.filter(created_at__date__gte=month_start) elif date_range == 'last_30_days': last_30_days = today - datetime.timedelta(days=30) queryset = queryset.filter(created_at__date__gte=last_30_days) # Handle custom date range if date_from: try: from_date = datetime.datetime.strptime(date_from, '%Y-%m-%d').date() queryset = queryset.filter(created_at__date__gte=from_date) except ValueError: pass if date_to: try: to_date = datetime.datetime.strptime(date_to, '%Y-%m-%d').date() queryset = queryset.filter(created_at__date__lte=to_date) except ValueError: pass return queryset def _apply_sorting(self, queryset, request): """ Apply sorting to queryset. Args: queryset: Base queryset request: Django HTTP request object Returns: QuerySet: Sorted queryset """ sort_by = request.GET.get('sort_by', 'name') valid_sort_fields = [ 'name', '-name', 'legal_name', '-legal_name', 'agency_type__name', '-agency_type__name', 'founded_year', '-founded_year', 'employee_count', '-employee_count', 'is_verified', '-is_verified', 'is_featured', '-is_featured', 'created_at', '-created_at', 'updated_at', '-updated_at' ] if sort_by in valid_sort_fields: queryset = queryset.order_by(sort_by) else: queryset = queryset.order_by('name') return queryset def _build_context(self, request: HttpRequest, page_obj, paginator) -> Dict[str, Any]: """ Build template context with all necessary data. Args: request: Django HTTP request object page_obj: Paginated page object paginator: Paginator instance Returns: dict: Template context """ user = request.user context = { # Pagination data 'agencies': page_obj.object_list, 'page_obj': page_obj, 'paginator': paginator, 'is_paginated': paginator.num_pages > 1, # Filter data 'agency_types': self._get_agency_types(), 'status_choices': self._get_status_choices(), 'size_choices': self._get_size_choices(), 'filter_params': self._get_current_filters(request), 'sort_options': self._get_sort_options(), # Statistics 'stats': self._get_stats(user), # Additional data 'date_range': self._get_date_range(), 'total_results': paginator.count, 'current_page': page_obj.number, 'total_pages': paginator.num_pages, # Filter states for template 'search_query': request.GET.get('search', ''), 'selected_agency_type': request.GET.get('agency_type', ''), 'selected_status': request.GET.get('status', ''), 'selected_size': request.GET.get('size', ''), } # Add extra context context.update(self.extra_context) return context def _get_agency_types(self): """ Get available agency types for dropdown filter. Returns: QuerySet: AgencyType objects ordered by sort_order and name """ try: return AgencyType.objects.all() except Exception: # Fallback if AgencyType model is not available try: AgencyType.create_default_types() return AgencyType.objects.filter(is_deleted=False).order_by('sort_order', 'name') except: return [] def _get_status_choices(self): """ Get available status choices for dropdown filter. Returns: list: Status choice tuples """ return [ ('all', _('All Status')), ('active', _('Active')), ('inactive', _('Inactive')), ('verified', _('Verified')), ('featured', _('Featured')), ('deleted', _('Deleted')), ] def _get_size_choices(self): """ Get available agency size choices for dropdown filter. Returns: list: Agency size choice tuples """ return [ ('all', _('All Sizes')), ('startup', _('Startup (1-10 employees)')), ('small', _('Small (11-50 employees)')), ('medium', _('Medium (51-200 employees)')), ('large', _('Large (201-1000 employees)')), ('enterprise', _('Enterprise (1000+ employees)')), ] def _get_current_filters(self, request: HttpRequest) -> Dict[str, str]: """ Get current filter parameters for template. Args: request: Django HTTP request object Returns: dict: Current filter values """ return { 'search': request.GET.get('search', ''), 'agency_type': request.GET.get('agency_type', 'all'), 'status': request.GET.get('status', 'all'), 'size': request.GET.get('size', 'all'), 'date_from': request.GET.get('date_from', ''), 'date_to': request.GET.get('date_to', ''), 'date_range': request.GET.get('date_range', ''), 'sort_by': request.GET.get('sort_by', 'name'), 'per_page': request.GET.get('per_page', '20'), } def _get_sort_options(self) -> list: """ Get available sorting options for dropdown. Returns: list: Sort options for template """ return [ {'value': 'name', 'label': _('Name (A-Z)')}, {'value': '-name', 'label': _('Name (Z-A)')}, {'value': 'legal_name', 'label': _('Legal Name (A-Z)')}, {'value': '-legal_name', 'label': _('Legal Name (Z-A)')}, {'value': 'agency_type__name', 'label': _('Type (A-Z)')}, {'value': '-agency_type__name', 'label': _('Type (Z-A)')}, {'value': '-founded_year', 'label': _('Founded (Newest)')}, {'value': 'founded_year', 'label': _('Founded (Oldest)')}, {'value': '-employee_count', 'label': _('Size (Largest)')}, {'value': 'employee_count', 'label': _('Size (Smallest)')}, {'value': '-is_verified', 'label': _('Verified First')}, {'value': '-is_featured', 'label': _('Featured First')}, {'value': '-created_at', 'label': _('Newest First')}, {'value': 'created_at', 'label': _('Oldest First')}, {'value': '-updated_at', 'label': _('Recently Updated')}, ] def _get_stats(self, user) -> Dict[str, int]: """ Get agencies statistics for dashboard. Args: user: Current user object Returns: dict: Agencies statistics """ # Base queryset with user scoping base_queryset = Agency.objects.all() if not user.is_superuser: base_queryset = base_queryset.filter(owner=user) today = timezone.now().date() week_ago = today - datetime.timedelta(days=7) return { 'total_agencies': base_queryset.count(), 'active_agencies': base_queryset.filter(is_active=True, is_deleted=False).count(), 'inactive_agencies': base_queryset.filter(is_active=False, is_deleted=False).count(), 'verified_agencies': base_queryset.filter(is_verified=True, is_deleted=False).count(), 'featured_agencies': base_queryset.filter(is_featured=True, is_deleted=False).count(), 'deleted_agencies': base_queryset.filter(is_deleted=True).count(), 'recent_agencies': base_queryset.filter( created_at__date__gte=week_ago, is_deleted=False ).count(), 'agencies_by_type': base_queryset.filter(is_deleted=False).values( 'agency_type__name' ).annotate(count=Count('id')).order_by('-count'), 'agencies_by_size': base_queryset.filter(is_deleted=False).values( 'agency_size' ).annotate(count=Count('id')).order_by('-count'), 'premium_types_count': base_queryset.filter( agency_type__is_premium=True, is_deleted=False ).count(), } def _get_date_range(self) -> Dict[str, datetime.date]: """ Get date range information for filtering. Returns: dict: Date range options """ today = timezone.now().date() return { 'today': today, 'yesterday': today - datetime.timedelta(days=1), 'week_start': today - datetime.timedelta(days=today.weekday()), 'month_start': today.replace(day=1), 'last_30_days': today - datetime.timedelta(days=30), } def _get_items_per_page(self, request: HttpRequest) -> int: """ Get number of items per page from request. Args: request: Django HTTP request object Returns: int: Number of items per page (default 20, max 100) """ try: per_page = int(request.GET.get('per_page', 20)) return max(10, min(per_page, 100)) # Between 10 and 100 except (ValueError, TypeError): return 20 def _log_user_activity(self, request: HttpRequest): """ Log user activity for auditing purposes. Args: request: Django HTTP request object """ try: Activity.log_activity( user=request.user, action="view_agencies_list", description=f"User {request.user.email if request.user else 'System'} viewed agencies list", request=request, ip_address=get_client_ip(request), user_agent=request.META.get("HTTP_USER_AGENT", ""), metadata={ 'search': request.GET.get('search', ''), 'agency_type': request.GET.get('agency_type', ''), 'status': request.GET.get('status', ''), 'size': request.GET.get('size', ''), 'sort_by': request.GET.get('sort_by', 'name'), 'page': request.GET.get('page', 1), 'per_page': request.GET.get('per_page', 20), } ) except Exception as e: # Log activity failure shouldn't break the view print(f"Failed to log activity: {e}") def _handle_ajax_request(self, request: HttpRequest) -> JsonResponse: """ Handle AJAX requests for dynamic updates. Args: request: Django HTTP request object Returns: JsonResponse: JSON response with agencies data """ try: # Get filtered queryset queryset = self._get_filtered_queryset(request) # Apply pagination paginator = Paginator(queryset, self._get_items_per_page(request)) page_number = request.GET.get('page', 1) page_obj = paginator.get_page(page_number) # Build agencies data agencies_data = [] for agency in page_obj.object_list: agency_data = { 'id': str(agency.id), 'name': agency.name, 'legal_name': agency.legal_name or '', 'description': agency.description or '', 'email': agency.email, 'phone': agency.phone or '', 'website': agency.website or '', 'agency_type': { 'id': agency.agency_type.id if agency.agency_type else None, 'name': agency.agency_type.name if agency.agency_type else '', 'color_code': agency.agency_type.color_code if agency.agency_type else '#007bff', 'is_premium': agency.agency_type.is_premium if agency.agency_type else False, }, 'is_active': agency.is_active, 'is_verified': agency.is_verified, 'is_featured': agency.is_featured, 'agency_size': agency.agency_size, 'employee_count': agency.employee_count, 'founded_year': agency.founded_year, 'created_at': agency.created_at.isoformat() if agency.created_at else None, 'logo_url': agency.logo.url if agency.logo else None, } agencies_data.append(agency_data) return JsonResponse({ 'success': True, 'agencies': agencies_data, 'total_results': paginator.count, 'current_page': page_obj.number, 'total_pages': paginator.num_pages, 'has_previous': page_obj.has_previous(), 'has_next': page_obj.has_next(), 'previous_page_number': page_obj.previous_page_number() if page_obj.has_previous() else None, 'next_page_number': page_obj.next_page_number() if page_obj.has_next() else None, }) except Exception as e: return JsonResponse({ 'success': False, 'error': str(e) }, status=500) class AgencyDetailView(LoginRequiredMixin, View): """Agency Detail View with related brands summary. Template: agencies/agency/detail.html """ http_method_names = ["get"] template_name = "agencies/agency/detail.html" def get(self, request: HttpRequest, pk: str) -> HttpResponse: agency = get_object_or_404(Agency, pk=pk) if not request.user.is_superuser and agency.owner != request.user: raise Http404("Agency not found") brands = Brand.objects.filter(agency=agency).select_related('agency').order_by('name') Activity.log_activity( user=request.user, action="view_agency_detail_cbv", description=f"Viewed agency detail (CBV): {agency.name}", request=request, ip_address=get_client_ip(request), user_agent=request.META.get("HTTP_USER_AGENT", ""), metadata={'agency_id': str(agency.id)} ) context = { 'agency': agency, 'brands': brands, 'title': f'{agency.name} - Details', 'active_menu': 'agencies', } return render(request, self.template_name, context) # Helper function for agency forms def get_agency_types(): """ Helper function to get agency types for form dropdowns. Returns: QuerySet: AgencyType objects ordered by sort_order and name """ try: return AgencyType.objects.filter(is_deleted=False).order_by('sort_order', 'name') except Exception: # Fallback if AgencyType model is not available return [] class AgencyCreateView(LoginRequiredMixin, View): """ Create a new Agency with comprehensive form handling. This view handles all agency form fields including: - Basic information (name, legal name, description) - Contact information (email, phone, website) - Business information (agency type, size, employee count, specializations) - Address information (street, city, state, country, postal code) - Localization (time zone, language) - Logo upload Template: agencies/agency/form.html """ http_method_names = ["get", "post"] template_name = "agencies/agency/form.html" def get(self, request: HttpRequest) -> HttpResponse: """ Render the agency creation form with all necessary dropdowns populated. Args: request: HTTP request object Returns: HttpResponse: Rendered form template """ # Get agency types for dropdown agency_types = self._get_agency_types() # Get choices for dropdowns size_choices = self._get_size_choices() specialization_choices = self._get_specialization_choices() timezone_choices = self._get_timezone_choices() language_choices = self._get_language_choices() country_choices = self._get_country_choices() context = { 'form_title': 'Create New Agency', 'submit_text': 'Create Agency', 'is_create': True, 'agency_types': agency_types, 'size_choices': size_choices, 'specialization_choices': specialization_choices, 'timezone_choices': timezone_choices, 'language_choices': language_choices, 'country_choices': country_choices, 'title': 'Create Agency - Focus', 'agency': EmptyAgency(), 'active_menu': 'agencies', } return render(request, self.template_name, context) def post(self, request: HttpRequest) -> HttpResponse: """ Handle agency creation with comprehensive validation and error handling. Args: request: HTTP request object Returns: HttpResponse: Redirect to agency detail or form with errors """ try: # Extract and validate form data form_data = self._extract_form_data(request) validation_errors = self._validate_form_data(form_data, request) if validation_errors: # Return form with errors return self._render_form_with_errors(request, validation_errors, form_data) # Create the agency agency = self._create_agency(form_data, request) # Handle logo upload if provided if 'logo' in request.FILES: self._handle_logo_upload(agency, request.FILES['logo']) # Log the activity self._log_activity(request, agency, 'create') # Success message and redirect messages.success(request, f'Agency "{agency.name}" has been created successfully!') return redirect('agencies:agency_detail', pk=agency.pk) except ValidationError as e: messages.error(request, f'Validation error: {str(e)}') return self.get(request) except Exception as e: messages.error(request, f'An unexpected error occurred: {str(e)}') return self.get(request) def _extract_form_data(self, request: HttpRequest) -> dict: """ Extract all form data from the request. Args: request: HTTP request object Returns: dict: Extracted form data """ return { # Basic information 'name': request.POST.get('name', '').strip(), 'legal_name': request.POST.get('legal_name', '').strip(), 'description': request.POST.get('description', '').strip(), 'agency_type_id': request.POST.get('agency_type', '').strip(), # Contact information 'email': request.POST.get('email', '').strip(), 'phone': request.POST.get('phone', '').strip(), 'website': request.POST.get('website', '').strip(), 'founded_year': request.POST.get('founded_year', '').strip(), # Business information 'agency_size': request.POST.get('agency_size', '').strip(), 'employee_count': request.POST.get('employee_count', '').strip(), 'time_zone': request.POST.get('time_zone', 'UTC').strip(), 'language': request.POST.get('language', 'en').strip(), 'specializations': request.POST.getlist('specializations'), # Address information 'street_address': request.POST.get('street_address', '').strip(), 'city': request.POST.get('city', '').strip(), 'state': request.POST.get('state', '').strip(), 'postal_code': request.POST.get('postal_code', '').strip(), 'country': request.POST.get('country', '').strip(), } def _validate_form_data(self, form_data: dict, request: HttpRequest) -> list: """ Validate all form data and return list of errors. Args: form_data: Extracted form data request: HTTP request object Returns: list: List of validation error messages """ errors = [] # Required field validation if not form_data['name']: errors.append('Agency name is required.') elif len(form_data['name']) > 200: errors.append('Agency name cannot exceed 200 characters.') if not form_data['email']: errors.append('Email address is required.') elif not self._is_valid_email(form_data['email']): errors.append('Please enter a valid email address.') if not form_data['agency_type_id']: errors.append('Agency type is required.') else: # Validate agency type exists try: AgencyType.objects.get(id=form_data['agency_type_id']) except AgencyType.DoesNotExist: errors.append('Selected agency type is invalid.') # Optional field validation if form_data['legal_name'] and len(form_data['legal_name']) > 200: errors.append('Legal name cannot exceed 200 characters.') if form_data['description'] and len(form_data['description']) > 2000: errors.append('Description cannot exceed 2000 characters.') if form_data['website'] and not self._is_valid_url(form_data['website']): errors.append('Please enter a valid website URL.') if form_data['phone'] and len(form_data['phone']) > 20: errors.append('Phone number cannot exceed 20 characters.') # Validate founded year if form_data['founded_year']: try: year = int(form_data['founded_year']) current_year = timezone.now().year if year < 1900 or year > current_year + 1: errors.append(f'Founded year must be between 1900 and {current_year + 1}.') except ValueError: errors.append('Founded year must be a valid number.') # Validate employee count if form_data['employee_count']: try: count = int(form_data['employee_count']) if count < 1 or count > 1000000: errors.append('Employee count must be between 1 and 1,000,000.') except ValueError: errors.append('Employee count must be a valid number.') # Validate address fields length if form_data['street_address'] and len(form_data['street_address']) > 255: errors.append('Street address cannot exceed 255 characters.') if form_data['city'] and len(form_data['city']) > 100: errors.append('City cannot exceed 100 characters.') if form_data['state'] and len(form_data['state']) > 100: errors.append('State cannot exceed 100 characters.') if form_data['postal_code'] and len(form_data['postal_code']) > 20: errors.append('Postal code cannot exceed 20 characters.') # Validate logo file if uploaded if 'logo' in request.FILES: logo_file = request.FILES['logo'] logo_errors = self._validate_logo_file(logo_file) errors.extend(logo_errors) # Check for duplicate agency name for this user if form_data['name']: existing = Agency.objects.filter( name__iexact=form_data['name'], owner=request.user, is_deleted=False ).exists() if existing: errors.append('You already have an agency with this name.') return errors def _validate_logo_file(self, logo_file) -> list: """ Validate uploaded logo file. Args: logo_file: Uploaded file object Returns: list: List of validation errors """ errors = [] # Check file size (5MB limit) max_size = 5 * 1024 * 1024 # 5MB if logo_file.size > max_size: errors.append('Logo file size cannot exceed 5MB.') # Check file extension allowed_extensions = ['jpg', 'jpeg', 'png', 'svg', 'webp'] file_extension = logo_file.name.split('.')[-1].lower() if '.' in logo_file.name else '' if file_extension not in allowed_extensions: errors.append('Logo must be in JPG, JPEG, PNG, SVG, or WebP format.') return errors def _create_agency(self, form_data: dict, request: HttpRequest): """ Create the agency with validated data. Args: form_data: Validated form data request: HTTP request object Returns: Agency: Created agency instance """ # Get agency type agency_type = AgencyType.objects.get(id=form_data['agency_type_id']) # Prepare agency data agency_data = { 'name': form_data['name'], 'owner': request.user, 'agency_type': agency_type, 'email': form_data['email'], 'is_active': True, # New agencies are active by default } # Add optional fields if provided optional_fields = [ 'legal_name', 'description', 'phone', 'website', 'agency_size', 'time_zone', 'language', 'street_address', 'city', 'state', 'postal_code', 'country' ] for field in optional_fields: if form_data[field]: agency_data[field] = form_data[field] # Handle numeric fields if form_data['founded_year']: agency_data['founded_year'] = int(form_data['founded_year']) if form_data['employee_count']: agency_data['employee_count'] = int(form_data['employee_count']) # Handle specializations (JSON field) if form_data['specializations']: agency_data['specializations'] = form_data['specializations'] # Create the agency agency = Agency.objects.create(**agency_data) return agency def _handle_logo_upload(self, agency, logo_file): """ Handle logo file upload. Args: agency: Agency instance logo_file: Uploaded logo file """ agency.logo = logo_file agency.save() def _render_form_with_errors(self, request: HttpRequest, errors: list, form_data: dict) -> HttpResponse: """ Render form with validation errors and preserve user input. Args: request: HTTP request object errors: List of error messages form_data: Form data to preserve Returns: HttpResponse: Form with errors """ # Add errors to messages for error in errors: messages.error(request, error) # Get dropdown data agency_types = self._get_agency_types() size_choices = self._get_size_choices() specialization_choices = self._get_specialization_choices() timezone_choices = self._get_timezone_choices() language_choices = self._get_language_choices() country_choices = self._get_country_choices() context = { 'form_title': 'Create New Agency', 'submit_text': 'Create Agency', 'is_create': True, 'agency_types': agency_types, 'size_choices': size_choices, 'specialization_choices': specialization_choices, 'timezone_choices': timezone_choices, 'language_choices': language_choices, 'country_choices': country_choices, 'form_data': form_data, # Preserve form data 'agency': EmptyAgency(), 'title': 'Create Agency - Focus', 'active_menu': 'agencies', } return render(request, self.template_name, context) def _log_activity(self, request: HttpRequest, agency, action: str): """ Log user activity. Args: request: HTTP request object agency: Agency instance action: Action type ('create', 'update', etc.) """ try: Activity.log_activity( user=request.user, action=f"{action}_agency_cbv", description=f"User {request.user.email} {action}d agency: {agency.name}", request=request, ip_address=get_client_ip(request), user_agent=request.META.get("HTTP_USER_AGENT", ""), metadata={ 'agency_id': str(agency.id), 'agency_name': agency.name, 'action': action } ) except Exception as e: # Log activity failure shouldn't break the view print(f"Failed to log activity: {e}") # Helper methods for dropdown data def _get_agency_types(self): """Get agency types for dropdown.""" try: return AgencyType.objects.filter(is_deleted=False).order_by('sort_order', 'name') except Exception: # If no agency types exist, create defaults try: AgencyType.create_default_types() return AgencyType.objects.filter(is_deleted=False).order_by('sort_order', 'name') except: return [] def _get_size_choices(self): """Get agency size choices.""" return [ ('', 'Select agency size...'), ('startup', 'Startup (1-10 employees)'), ('small', 'Small (11-50 employees)'), ('medium', 'Medium (51-200 employees)'), ('large', 'Large (201-1000 employees)'), ('enterprise', 'Enterprise (1000+ employees)'), ] def _get_specialization_choices(self): """Get specialization choices.""" return [ ('tv_advertising', 'TV Advertising'), ('digital_marketing', 'Digital Marketing'), ('creative_services', 'Creative Services'), ('media_planning', 'Media Planning'), ('brand_strategy', 'Brand Strategy'), ('public_relations', 'Public Relations'), ('social_media', 'Social Media'), ('content_creation', 'Content Creation'), ('data_analytics', 'Data Analytics'), ('full_service', 'Full Service'), ] def _get_timezone_choices(self): """Get timezone choices.""" return [ ('UTC', 'UTC'), ('America/New_York', 'Eastern Time'), ('America/Chicago', 'Central Time'), ('America/Denver', 'Mountain Time'), ('America/Los_Angeles', 'Pacific Time'), ('Europe/London', 'London Time'), ('Europe/Paris', 'Paris Time'), ('Europe/Berlin', 'Berlin Time'), ('Asia/Tokyo', 'Tokyo Time'), ('Australia/Sydney', 'Sydney Time'), ] def _get_language_choices(self): """Get language choices.""" return [ ('en', 'English'), ('es', 'Spanish'), ('fr', 'French'), ('de', 'German'), ('it', 'Italian'), ('pt', 'Portuguese'), ('zh', 'Chinese'), ('ja', 'Japanese'), ('ko', 'Korean'), ('ar', 'Arabic'), ] def _get_country_choices(self): """Get country choices.""" return [ ('', 'Select country...'), ('US', 'United States'), ('CA', 'Canada'), ('GB', 'United Kingdom'), ('DE', 'Germany'), ('FR', 'France'), ('ES', 'Spain'), ('IT', 'Italy'), ('AU', 'Australia'), ('JP', 'Japan'), ('CN', 'China'), ('IN', 'India'), ('BR', 'Brazil'), ('MX', 'Mexico'), ('NL', 'Netherlands'), ('SE', 'Sweden'), ('NO', 'Norway'), ('DK', 'Denmark'), ('FI', 'Finland'), ] def _is_valid_email(self, email: str) -> bool: """Validate email format.""" import re pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$' return re.match(pattern, email) is not None def _is_valid_url(self, url: str) -> bool: """Validate URL format.""" try: from urllib.parse import urlparse result = urlparse(url) return all([result.scheme, result.netloc]) except Exception: return False def post(self, request: HttpRequest) -> HttpResponse: """ Handle agency creation with comprehensive validation and error handling. Args: request: HTTP request object Returns: HttpResponse: Redirect to agency detail or form with errors """ try: # Extract and validate form data form_data = self._extract_form_data(request) validation_errors = self._validate_form_data(form_data, request) if validation_errors: # Return form with errors return self._render_form_with_errors(request, validation_errors, form_data) # Create the agency agency = self._create_agency(form_data, request) # Handle logo upload if provided if 'logo' in request.FILES: self._handle_logo_upload(agency, request.FILES['logo']) # Log the activity self._log_activity(request, agency, 'create') # Success message and redirect messages.success(request, f'Agency "{agency.name}" has been created successfully!') return redirect('agencies:agency_detail', pk=agency.pk) except ValidationError as e: messages.error(request, f'Validation error: {str(e)}') return self.get(request) except Exception as e: messages.error(request, f'An unexpected error occurred: {str(e)}') return self.get(request) def _extract_form_data(self, request: HttpRequest) -> dict: """ Extract all form data from the request. Args: request: HTTP request object Returns: dict: Extracted form data """ return { # Basic information 'name': request.POST.get('name', '').strip(), 'legal_name': request.POST.get('legal_name', '').strip(), 'description': request.POST.get('description', '').strip(), 'agency_type_id': request.POST.get('agency_type', '').strip(), # Contact information 'email': request.POST.get('email', '').strip(), 'phone': request.POST.get('phone', '').strip(), 'website': request.POST.get('website', '').strip(), 'founded_year': request.POST.get('founded_year', '').strip(), # Business information 'agency_size': request.POST.get('agency_size', '').strip(), 'employee_count': request.POST.get('employee_count', '').strip(), 'time_zone': request.POST.get('time_zone', 'UTC').strip(), 'language': request.POST.get('language', 'en').strip(), 'specializations': request.POST.getlist('specializations'), # Address information 'street_address': request.POST.get('street_address', '').strip(), 'city': request.POST.get('city', '').strip(), 'state': request.POST.get('state', '').strip(), 'postal_code': request.POST.get('postal_code', '').strip(), 'country': request.POST.get('country', '').strip(), } def _validate_form_data(self, form_data: dict, request: HttpRequest) -> list: """ Validate all form data and return list of errors. Args: form_data: Extracted form data request: HTTP request object Returns: list: List of validation error messages """ errors = [] # Required field validation if not form_data['name']: errors.append('Agency name is required.') elif len(form_data['name']) > 200: errors.append('Agency name cannot exceed 200 characters.') if not form_data['email']: errors.append('Email address is required.') elif not self._is_valid_email(form_data['email']): errors.append('Please enter a valid email address.') if not form_data['agency_type_id']: errors.append('Agency type is required.') else: # Validate agency type exists try: AgencyType.objects.get(id=form_data['agency_type_id']) except AgencyType.DoesNotExist: errors.append('Selected agency type is invalid.') # Optional field validation if form_data['legal_name'] and len(form_data['legal_name']) > 200: errors.append('Legal name cannot exceed 200 characters.') if form_data['description'] and len(form_data['description']) > 2000: errors.append('Description cannot exceed 2000 characters.') if form_data['website'] and not self._is_valid_url(form_data['website']): errors.append('Please enter a valid website URL.') if form_data['phone'] and len(form_data['phone']) > 20: errors.append('Phone number cannot exceed 20 characters.') # Validate founded year if form_data['founded_year']: try: year = int(form_data['founded_year']) current_year = timezone.now().year if year < 1900 or year > current_year + 1: errors.append(f'Founded year must be between 1900 and {current_year + 1}.') except ValueError: errors.append('Founded year must be a valid number.') # Validate employee count if form_data['employee_count']: try: count = int(form_data['employee_count']) if count < 1 or count > 1000000: errors.append('Employee count must be between 1 and 1,000,000.') except ValueError: errors.append('Employee count must be a valid number.') # Validate address fields length if form_data['street_address'] and len(form_data['street_address']) > 255: errors.append('Street address cannot exceed 255 characters.') if form_data['city'] and len(form_data['city']) > 100: errors.append('City cannot exceed 100 characters.') if form_data['state'] and len(form_data['state']) > 100: errors.append('State cannot exceed 100 characters.') if form_data['postal_code'] and len(form_data['postal_code']) > 20: errors.append('Postal code cannot exceed 20 characters.') # Validate logo file if uploaded if 'logo' in request.FILES: logo_file = request.FILES['logo'] logo_errors = self._validate_logo_file(logo_file) errors.extend(logo_errors) # Check for duplicate agency name for this user if form_data['name']: existing = Agency.objects.filter( name__iexact=form_data['name'], owner=request.user, is_deleted=False ).exists() if existing: errors.append('You already have an agency with this name.') return errors def _validate_logo_file(self, logo_file) -> list: """ Validate uploaded logo file. Args: logo_file: Uploaded file object Returns: list: List of validation errors """ errors = [] # Check file size (5MB limit) max_size = 5 * 1024 * 1024 # 5MB if logo_file.size > max_size: errors.append('Logo file size cannot exceed 5MB.') # Check file extension allowed_extensions = ['jpg', 'jpeg', 'png', 'svg', 'webp'] file_extension = logo_file.name.split('.')[-1].lower() if '.' in logo_file.name else '' if file_extension not in allowed_extensions: errors.append('Logo must be in JPG, JPEG, PNG, SVG, or WebP format.') return errors def _create_agency(self, form_data: dict, request: HttpRequest): """ Create the agency with validated data. Args: form_data: Validated form data request: HTTP request object Returns: Agency: Created agency instance """ # Get agency type agency_type = AgencyType.objects.get(id=form_data['agency_type_id']) # Prepare agency data agency_data = { 'name': form_data['name'], 'owner': request.user, 'agency_type': agency_type, 'email': form_data['email'], 'is_active': True, # New agencies are active by default } # Add optional fields if provided optional_fields = [ 'legal_name', 'description', 'phone', 'website', 'agency_size', 'time_zone', 'language', 'street_address', 'city', 'state', 'postal_code', 'country' ] for field in optional_fields: if form_data[field]: agency_data[field] = form_data[field] # Handle numeric fields if form_data['founded_year']: agency_data['founded_year'] = int(form_data['founded_year']) if form_data['employee_count']: agency_data['employee_count'] = int(form_data['employee_count']) # Handle specializations (JSON field) if form_data['specializations']: agency_data['specializations'] = form_data['specializations'] # Create the agency agency = Agency.objects.create(**agency_data) return agency def _handle_logo_upload(self, agency, logo_file): """ Handle logo file upload. Args: agency: Agency instance logo_file: Uploaded logo file """ agency.logo = logo_file agency.save() def _render_form_with_errors(self, request: HttpRequest, errors: list, form_data: dict) -> HttpResponse: """ Render form with validation errors and preserve user input. Args: request: HTTP request object errors: List of error messages form_data: Form data to preserve Returns: HttpResponse: Form with errors """ # Add errors to messages for error in errors: messages.error(request, error) # Get dropdown data agency_types = self._get_agency_types() size_choices = self._get_size_choices() specialization_choices = self._get_specialization_choices() timezone_choices = self._get_timezone_choices() language_choices = self._get_language_choices() country_choices = self._get_country_choices() context = { 'form_title': 'Create New Agency', 'submit_text': 'Create Agency', 'is_create': True, 'agency_types': agency_types, 'size_choices': size_choices, 'specialization_choices': specialization_choices, 'timezone_choices': timezone_choices, 'language_choices': language_choices, 'country_choices': country_choices, 'form_data': form_data, # Preserve form data 'agency': EmptyAgency(), 'title': 'Create Agency - Focus', 'active_menu': 'agencies', } return render(request, self.template_name, context) def _log_activity(self, request: HttpRequest, agency, action: str): """ Log user activity. Args: request: HTTP request object agency: Agency instance action: Action type ('create', 'update', etc.) """ try: Activity.log_activity( user=request.user, action=f"{action}_agency_cbv", description=f"User {request.user.email} {action}d agency: {agency.name}", request=request, ip_address=get_client_ip(request), user_agent=request.META.get("HTTP_USER_AGENT", ""), metadata={ 'agency_id': str(agency.id), 'agency_name': agency.name, 'action': action } ) except Exception as e: # Log activity failure shouldn't break the view print(f"Failed to log activity: {e}") # Helper methods for dropdown data def _get_agency_types(self): """Get agency types for dropdown.""" try: return AgencyType.objects.filter(is_deleted=False).order_by('sort_order', 'name') except Exception: # If no agency types exist, create defaults try: AgencyType.create_default_types() return AgencyType.objects.filter(is_deleted=False).order_by('sort_order', 'name') except: return [] def _get_size_choices(self): """Get agency size choices.""" return [ ('', 'Select agency size...'), ('startup', 'Startup (1-10 employees)'), ('small', 'Small (11-50 employees)'), ('medium', 'Medium (51-200 employees)'), ('large', 'Large (201-1000 employees)'), ('enterprise', 'Enterprise (1000+ employees)'), ] def _get_specialization_choices(self): """Get specialization choices.""" return [ ('tv_advertising', 'TV Advertising'), ('digital_marketing', 'Digital Marketing'), ('creative_services', 'Creative Services'), ('media_planning', 'Media Planning'), ('brand_strategy', 'Brand Strategy'), ('public_relations', 'Public Relations'), ('social_media', 'Social Media'), ('content_creation', 'Content Creation'), ('data_analytics', 'Data Analytics'), ('full_service', 'Full Service'), ] def _get_timezone_choices(self): """Get timezone choices.""" return [ ('UTC', 'UTC'), ('America/New_York', 'Eastern Time'), ('America/Chicago', 'Central Time'), ('America/Denver', 'Mountain Time'), ('America/Los_Angeles', 'Pacific Time'), ('Europe/London', 'London Time'), ('Europe/Paris', 'Paris Time'), ('Europe/Berlin', 'Berlin Time'), ('Asia/Tokyo', 'Tokyo Time'), ('Australia/Sydney', 'Sydney Time'), ] def _get_language_choices(self): """Get language choices.""" return [ ('en', 'English'), ('es', 'Spanish'), ('fr', 'French'), ('de', 'German'), ('it', 'Italian'), ('pt', 'Portuguese'), ('zh', 'Chinese'), ('ja', 'Japanese'), ('ko', 'Korean'), ('ar', 'Arabic'), ] def _get_country_choices(self): """Get country choices.""" return [ ('', 'Select country...'), ('US', 'United States'), ('CA', 'Canada'), ('GB', 'United Kingdom'), ('DE', 'Germany'), ('FR', 'France'), ('ES', 'Spain'), ('IT', 'Italy'), ('AU', 'Australia'), ('JP', 'Japan'), ('CN', 'China'), ('IN', 'India'), ('BR', 'Brazil'), ('MX', 'Mexico'), ('NL', 'Netherlands'), ('SE', 'Sweden'), ('NO', 'Norway'), ('DK', 'Denmark'), ('FI', 'Finland'), ] def _is_valid_email(self, email: str) -> bool: """Validate email format.""" import re pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$' return re.match(pattern, email) is not None def _is_valid_url(self, url: str) -> bool: """Validate URL format.""" try: from urllib.parse import urlparse result = urlparse(url) return all([result.scheme, result.netloc]) except Exception: return False class AgencyUpdateView(LoginRequiredMixin, View): """Update an existing Agency. Template: agencies/agency/form.html """ def get(self, request: HttpRequest, pk: str) -> HttpResponse: agency = get_object_or_404(Agency, pk=pk) if not request.user.is_superuser and agency.owner != request.user: raise Http404("Agency not found") return render(request, 'agencies/agency/form.html', { 'agency': agency, 'form_title': f'Update Agency: {agency.name}', 'submit_text': 'Update Agency', 'is_create': False, }) def post(self, request: HttpRequest, pk: str) -> HttpResponse: agency = get_object_or_404(Agency, pk=pk) if not request.user.is_superuser and agency.owner != request.user: raise Http404("Agency not found") name = (request.POST.get('name') or request.POST.get('name_agency') or '').strip() description = (request.POST.get('description') or request.POST.get('description_agency') or '').strip() if name: agency.name = name if description: agency.description = description agency.save() Activity.log_activity( user=request.user, action="update_agency_cbv", description=f"Updated agency (CBV): {agency.name}", request=request, ip_address=get_client_ip(request), user_agent=request.META.get("HTTP_USER_AGENT", ""), metadata={'agency_id': str(agency.id)} ) messages.success(request, f'Agency "{agency.name}" updated successfully.') return redirect('agencies:agency_detail', pk=agency.pk) class AgencyDeleteView(LoginRequiredMixin, View): """Soft-delete an Agency. Template: agencies/agency/confirm_delete.html """ def get(self, request: HttpRequest, pk: str) -> HttpResponse: agency = get_object_or_404(Agency, pk=pk) if not request.user.is_superuser and agency.owner != request.user: raise Http404("Agency not found") # Get dropdown data (same as create view) agency_types = self._get_agency_types() size_choices = self._get_size_choices() specialization_choices = self._get_specialization_choices() timezone_choices = self._get_timezone_choices() language_choices = self._get_language_choices() country_choices = self._get_country_choices() context = { 'agency': agency, 'form_title': f'Update Agency: {agency.name}', 'submit_text': 'Update Agency', 'is_create': False, 'agency_types': agency_types, 'size_choices': size_choices, 'specialization_choices': specialization_choices, 'timezone_choices': timezone_choices, 'language_choices': language_choices, 'country_choices': country_choices, 'title': f'Update {agency.name} - Focus', 'active_menu': 'agencies', } return render(request, 'agencies/agency/form.html', context) # Add helper methods for dropdown data (same as CreateView) def _get_agency_types(self): """Get agency types for dropdown.""" try: return AgencyType.objects.filter(is_deleted=False).order_by('sort_order', 'name') except Exception: try: AgencyType.create_default_types() return AgencyType.objects.filter(is_deleted=False).order_by('sort_order', 'name') except: return [] def _get_size_choices(self): """Get agency size choices.""" return [ ('', 'Select agency size...'), ('startup', 'Startup (1-10 employees)'), ('small', 'Small (11-50 employees)'), ('medium', 'Medium (51-200 employees)'), ('large', 'Large (201-1000 employees)'), ('enterprise', 'Enterprise (1000+ employees)'), ] def _get_specialization_choices(self): """Get specialization choices.""" return [ ('tv_advertising', 'TV Advertising'), ('digital_marketing', 'Digital Marketing'), ('creative_services', 'Creative Services'), ('media_planning', 'Media Planning'), ('brand_strategy', 'Brand Strategy'), ('public_relations', 'Public Relations'), ('social_media', 'Social Media'), ('content_creation', 'Content Creation'), ('data_analytics', 'Data Analytics'), ('full_service', 'Full Service'), ] def _get_timezone_choices(self): """Get timezone choices.""" return [ ('UTC', 'UTC'), ('America/New_York', 'Eastern Time'), ('America/Chicago', 'Central Time'), ('America/Denver', 'Mountain Time'), ('America/Los_Angeles', 'Pacific Time'), ('Europe/London', 'London Time'), ('Europe/Paris', 'Paris Time'), ('Europe/Berlin', 'Berlin Time'), ('Asia/Tokyo', 'Tokyo Time'), ('Australia/Sydney', 'Sydney Time'), ] def _get_language_choices(self): """Get language choices.""" return [ ('en', 'English'), ('es', 'Spanish'), ('fr', 'French'), ('de', 'German'), ('it', 'Italian'), ('pt', 'Portuguese'), ('zh', 'Chinese'), ('ja', 'Japanese'), ('ko', 'Korean'), ('ar', 'Arabic'), ] def _get_country_choices(self): """Get country choices.""" return [ ('', 'Select country...'), ('US', 'United States'), ('CA', 'Canada'), ('GB', 'United Kingdom'), ('DE', 'Germany'), ('FR', 'France'), ('ES', 'Spain'), ('IT', 'Italy'), ('AU', 'Australia'), ('JP', 'Japan'), ('CN', 'China'), ('IN', 'India'), ('BR', 'Brazil'), ('MX', 'Mexico'), ('NL', 'Netherlands'), ('SE', 'Sweden'), ('NO', 'Norway'), ('DK', 'Denmark'), ('FI', 'Finland'), ] def post(self, request: HttpRequest, pk: str) -> HttpResponse: agency = get_object_or_404(Agency, pk=pk) if not request.user.is_superuser and agency.owner != request.user: raise Http404("Agency not found") # Soft delete semantics consistent with existing FBV agency.datetime = timezone.now() agency.is_deleted = 1 agency.save() Activity.log_activity( user=request.user, action="delete_agency_cbv", description=f"Deleted (soft) agency (CBV): {agency.name}", request=request, ip_address=get_client_ip(request), user_agent=request.META.get("HTTP_USER_AGENT", ""), metadata={'agency_id': str(agency.id)} ) messages.success(request, f'Agency "{agency.name}" deleted successfully.') return redirect('agencies:agency_list')