U hU @s`dZddlZddlZddlZddlZddlmZmZddlmZm Z m Z m Z m Z ddl mZddlmZddlmZddlmZdd lmZdd lmZdd lmZmZmZmZdd lmZdd l m!Z!ddl"m#Z#eZ$ee%dddZ&ee%dddZ'e%e(dddZ)e(dddZ*eee%e fdddZ+ee%e fee%e fdddZ,e e-e%e e-e e%ee%dd d!Z.d=e e-e%e e-e e%ee-e(d#d$d%Z/d>e e-e e-e ee eee%e fd&d'd(Z0d?e e%e%d)d*d+Z1e%dd,d-Z2ee%e fee%e fd.d/d0Z3d@e-e-ee%e fd2d3d4Z4dAe-e-d6d7d8Z5dBe-e ee%e fd:d;d<Z6dS)Ca Adtlas Activities Utilities This module contains utility functions and helper classes for the activities app, providing common functionality for activity logging, data processing, and analysis. Features: - IP address and user agent extraction - Activity filtering and validation - Data export utilities - Performance monitoring - Security helpers - Analytics calculations Author: Adtlas Development Team Version: 1.0.0 Last Updated: 2025-01-27 N)datetime timedelta)DictListOptionalUnionAny)urlparse)settings)cache)timezone)get_user_model) ContentType)QCountAvgSum) HttpRequest)slugify)ValidationError)requestreturncCsddddddddg}|D]>}|j|}|rd|krD|dd }t|r|Sq|jdd S) a( Extract the client IP address from the request. This function handles various proxy configurations and header formats to accurately determine the client's IP address. Args: request: Django HttpRequest object Returns: str: Client IP address HTTP_X_FORWARDED_FORZHTTP_X_REAL_IPZHTTP_X_FORWARDEDZHTTP_X_CLUSTER_CLIENT_IPZHTTP_FORWARDED_FORZHTTP_FORWARDED REMOTE_ADDR,rz 127.0.0.1)METAgetsplitstrip is_valid_ip)rZ ip_headersheaderipr"0/var/www/html/Focus/src/apps/activities/utils.py get_client_ip,s    r$cCs"|jdd}|r|ddSdS)z Extract the user agent string from the request. Args: request: Django HttpRequest object Returns: str: User agent string (truncated to 500 characters) HTTP_USER_AGENTNUnknown)rr)r user_agentr"r"r#get_user_agentSs r*)r!rcCs,zt|WdStk r&YdSXdS)z Validate if a string is a valid IP address. Args: ip: IP address string to validate Returns: bool: True if valid IP address, False otherwise TFN) ipaddress ip_address ValueError)r!r"r"r#rcs  r)rc Cs|jj}|jj}ttdds dSttdddddg}||kr@dSttd d d d g}|d |}||krldSt|di}t|tr|ddSt|tr|Sttdddddddg}||kS)aT Determine if activities should be logged for a given model. This function checks various conditions to decide whether model changes should be automatically logged as activities. Args: model_class: Django model class Returns: bool: True if activities should be logged, False otherwise ZACTIVITIES_AUTO_LOG_ENABLEDTFZACTIVITIES_EXCLUDED_APPSsessionsadmin contenttypesauthZACTIVITIES_EXCLUDED_MODELSzactivities.activityzactivities.activitysummaryzcore.activitylog.Z_activity_loggingenabledZACTIVITIES_INCLUDED_APPSaccounts campaignschannels analyticsvast playlists) _meta app_label model_namegetattrr isinstancedictrbool) model_classr;r<Z excluded_appsZexcluded_modelsZ model_labelZ model_configZ included_appsr"r"r#should_log_activitytsD     rBcst|t||j|jtd}|jd}|r<||d<|j ri}ddddg}|j D]&\}t fdd |Ds\||<q\|r||d <|S) z Extract activity context information from the request. Args: request: Django HttpRequest object Returns: dict: Context information for activity logging )r,r)methodpath timestamp HTTP_REFERERrefererpasswordtokenkeysecretc3s|]}|kVqdSNlower).0excludedrJr"r# sz'get_activity_context.. query_params) r$r*rCrDr now isoformatrrGETitemsany)rcontextrGZ safe_paramsZexcluded_paramsvaluer"rQr#get_activity_contexts$     r[)metadatarcst|tsiSi}ddddddddd d d d d dg}|D]f\}tfdd|Drbd|<q:t|trzt||<q:t|trdd|D|<q:||<q:|S)z Sanitize metadata to remove sensitive information. Args: metadata: Raw metadata dictionary Returns: dict: Sanitized metadata rHpasswdpwdrKrJrIapi_keyZ access_tokenZ refresh_token session_key csrf_token auth_tokenZ private_keyZ public_keyc3s|]}|kVqdSrLrM)rOZ sensitiverQr"r#rRsz$sanitize_metadata..z [REDACTED]cSs"g|]}t|trt|n|qSr")r>r?sanitize_metadata)rOitemr"r"r# sz%sanitize_metadata..)r>r?rWrXrclist)r\Z sanitizedZsensitive_keysrZr"rQr#rcs6     rc)user_idactioncontent_type_id object_idrErc CsF|jddd}|d|d|d|d| }t|dS)aK Generate a unique hash for an activity to prevent duplicates. Args: user_id: User ID action: Activity action content_type_id: Content type ID object_id: Object ID timestamp: Activity timestamp (rounded to minute) Returns: str: SHA-256 hash of the activity r)second microsecond:zutf-8)replacerUhashlibsha256encode hexdigest)rgrhrirjrEZrounded_timestampZ hash_inputr"r"r#generate_activity_hashs$rs)rgrhrirjrEwindow_minutesrcCs>t|||||}d|}t|r(dSt|d|ddS)a Check if an activity is a duplicate within a time window. Args: user_id: User ID action: Activity action content_type_id: Content type ID object_id: Object ID timestamp: Activity timestamp window_minutes: Time window in minutes to check for duplicates Returns: bool: True if duplicate activity exists, False otherwise Zactivity_hash_T<F)rsr rset)rgrhrirjrEruZ activity_hash cache_keyr"r"r#is_duplicate_activitys  ry)rg category_id start_dateend_daterc Cs,ddlm}|j}|r&|j|d}|r6|j|d}|rF|j|d}|rV|j|d}|}|jdd}|jd d}|d kr||d nd } |jd d jtd ddpd } t| dj t dd ddd} |jd d d } | d } |||t| dt| d| | | dS)aV Calculate activity statistics for given parameters. Args: user_id: User ID to filter by (optional) category_id: Category ID to filter by (optional) start_date: Start date for filtering (optional) end_date: End date for filtering (optional) Returns: dict: Activity statistics rtActivity)rg)rz)created_at__gte)created_at__lteT is_successfulFrd)duration_ms__isnull duration_ms) avg_durationrrhcount-countN )Z user__isnulluserr,)total_activitiessuccessful_activitiesfailed_activities success_rateavg_duration_ms top_actions unique_users unique_ips)modelsr~objectsallfilterr aggregaterrfvaluesannotaterorder_bydistinctround)rgrzr{r|r~querysetrrrrrrrrr"r"r#get_activity_statistics2sR       r)filenamerc Cs ddl}ddlm}|}||}dddddd d d d d ddddg}|||dddD]}t|j|jrv|jj nd| |j |j r|j j nd|jrt|jnd|jpd|jpd|jpd|jrdnd|jpd|jpd|j|jrt|jndg}||q\|S)z Export activities to CSV format. Args: queryset: Activity queryset to export filename: Optional filename for the export Returns: str: CSV content as string rN)StringIOIDUserAction DescriptionCategoryz Content Typez Object IDz IP Addressz User Agentz Is Successfulz Duration (ms)z Error Messagez Created AtMetadatarcategory content_typeSystemr&YesNoz{})csviorwriterwriterowselect_relatedstridremailget_action_display descriptionrnamerrjr,r)rr error_message created_atrUr\jsondumpsgetvalue) rrrroutputrheadersactivityrowr"r"r#export_activities_to_csvwsL      rcCsg}|dddD]}t|j|jr,|jjnd|jr<|jjnd|jrN|jndd|j||j|j rp|j jnd|j r|j j nd|j r|j j ndd|j r|j jnd|j r|j j nd|j r|j jndd|j|j|j|j|j|j|j|j|jd}||qtj|d d d S) z Export activities to JSON format. Args: queryset: Activity queryset to export Returns: str: JSON content as string rrrN)rr full_name)rrcode)rr;model)rrrhZaction_displayrrrrjr,r)rrrr\r updated_atrF)indent ensure_ascii)rrrrr get_full_namerhrrrrrrr;rrjr,r)rrrr\rrUrappendrr)r activitiesr activity_datar"r"r#export_activities_to_jsons<  r)datarc Csi}|dsd|d<|ds(d|d<|dd}|rLtd|sLd|d<|d}|rjt|sjd |d<|d }|d k rz.t|}|d krd |d <n|dkrd|d <Wn ttfk rd|d <YnX|d}|d k rt|tsd|d<n t |}t |dkrd|d<|r"t || }|rr?rrlenrcopyrc)rerrorsrhr,rr\Z metadata_str cleaned_datar"r"r#validate_activity_datasP                  r)rgdaysrc CsNddlm}t}|t|d}|jj|||d}|}|jdd}|jdd}t| dd j t d d  d } t| d j t d d  d dd} g} t |D]>} |t| d} |j| d}| | |dqt|jddid dj t d d  d}||||||dkr>||dnd| | | |d S)z Get activity summary for a specific user. Args: user_id: User ID days: Number of days to include in summary Returns: dict: User activity summary rtr}r)rgrrTrFcategory__namecategory__coderrrrhNr)created_at__date)daterhourzEXTRACT(hour FROM created_at))selectrr) rg period_daysrrrrcategory_breakdownaction_breakdowndaily_activitieshourly_activities)rr~r rTrrrrrfrrrrrangerrrUextra)rgrr~r|r{rrrrrrridayZday_activitiesrr"r"r#get_user_activity_summary&st     rZ) days_to_keeprcCs8ddlm}tt|d}|jj|d\}}|S)z Clean up old activities based on retention policy. Args: days_to_keep: Number of days to keep activities Returns: int: Number of activities deleted rtr}r)created_at__lt)rr~r rTrrrdelete)rr~ cutoff_date deleted_count_r"r"r#cleanup_old_activitiesqs r)hoursrc Csddlm}g}t}|t|d}|jjd||ddjt ddjd d }|D]6}| d d |d|d d|d d|ddqV|jj||ddjt dt ddddjdd d}|D]H}| dd|d|d |dd|dd|d d|dddq|S) z Get security-related activity alerts. Args: hours: Number of hours to check for alerts Returns: list: List of security alerts rtr})r login_failed)rhrrr,rr) count__gteZmultiple_failed_loginshighrz Multiple failed login attempts (z ) from IP )typeseverityr,rmessage)rrrT)r)r user_countr)rZuser_count__gteZsuspicious_activitymediumrzSuspicious activity from IP z: z activities across z users)rrr,Zactivity_countrr) rr~r rTrrrrrrr) rr~Zalertsend_time start_timeZ failed_loginsZ failed_loginZsuspicious_ipsZ suspicious_ipr"r"r#get_security_alertssZ   " r)rt)NNNN)N)r)r)r)7__doc__rrror+rrtypingrrrrr urllib.parser django.confr django.core.cacher django.utilsr django.contrib.authr "django.contrib.contenttypes.modelsrdjango.db.modelsrrrr django.httprdjango.utils.textrdjango.core.exceptionsrrrr$r*r@rrBr[rcintrsryrrrrrrrr"r"r"r#sj         '8& $   " E10 NK