import calendar import json import math import re import jwt import datetime from django.conf import settings from django.contrib.admin.models import LogEntry, ADDITION from django.contrib.auth.models import User from django.contrib.contenttypes.models import ContentType from django.db import IntegrityError from django.http import JsonResponse from django.utils import timezone from django.views.decorators.csrf import csrf_exempt from django.db.models import Min, Max, Avg, Count, Sum, F, Value, CharField from django.db.models.functions import TruncMonth, Cast, Substr from django.views.decorators.http import require_POST, require_GET, require_http_methods from google.oauth2 import id_token from google.auth.transport import requests as google_requests from notas.models import Product, Order, OrderProduct, Vendor, UserProfile, Notification, ContaBasica, FamilyGroup, FamilyInvite, ImportJob from api.push import send_push_notification from api.spending import add_months, categorize_ncm, order_installment_amount_for_month # Module-level cache for Apple's JWKS client. Apple rotates these keys # periodically; PyJWKClient handles HTTP caching internally. _apple_jwks_client = None def _get_apple_jwks_client(): global _apple_jwks_client if _apple_jwks_client is None: _apple_jwks_client = jwt.PyJWKClient( 'https://appleid.apple.com/auth/keys', cache_keys=True, ) return _apple_jwks_client # --------------------------------------------------------------------------- # Decorators # --------------------------------------------------------------------------- def require_api_key(view_func): """Decorator that validates the X-API-Key header.""" def wrapper(request, *args, **kwargs): api_key = request.headers.get('X-API-Key', '') if api_key != settings.MOBILE_API_KEY: return JsonResponse( {'error': 'Chave de API inválida.'}, status=403, ) return view_func(request, *args, **kwargs) return wrapper def require_auth(view_func): """ Decorator that validates both X-API-Key AND the JWT Bearer token. On success, sets request.api_user to the Django User instance. """ def wrapper(request, *args, **kwargs): # 1. Validate API key api_key = request.headers.get('X-API-Key', '') if api_key != settings.MOBILE_API_KEY: return JsonResponse( {'error': 'Chave de API inválida.'}, status=403, ) # 2. Validate JWT (via X-Auth-Token header; avoids proxy stripping Authorization) token = request.headers.get('X-Auth-Token', '').strip() if not token: return JsonResponse( {'error': 'Token não fornecido.'}, status=401, ) payload = _decode_jwt(token) if payload is None: return JsonResponse( {'error': 'Token inválido ou expirado.'}, status=401, ) try: request.api_user = User.objects.get(id=payload['user_id']) except User.DoesNotExist: return JsonResponse( {'error': 'Usuário não encontrado.'}, status=404, ) return view_func(request, *args, **kwargs) return wrapper # --------------------------------------------------------------------------- # JWT helpers # --------------------------------------------------------------------------- def _generate_jwt(user): """Generate a JWT token for the given user.""" payload = { 'user_id': user.id, 'email': user.email, 'exp': datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(days=30), 'iat': datetime.datetime.now(datetime.timezone.utc), } return jwt.encode(payload, settings.SECRET_KEY, algorithm='HS256') def _decode_jwt(token): """Decode and validate a JWT token. Returns the payload or None.""" try: payload = jwt.decode(token, settings.SECRET_KEY, algorithms=['HS256']) return payload except (jwt.ExpiredSignatureError, jwt.InvalidTokenError): return None # --------------------------------------------------------------------------- # Serializers # --------------------------------------------------------------------------- def _user_to_dict(user): """Serialize a User object into a dict for the API response.""" return { 'id': user.id, 'email': user.email, 'name': user.get_full_name() or user.username, 'first_name': user.first_name, 'last_name': user.last_name, 'is_staff': user.is_staff, 'date_joined': user.date_joined.isoformat(), 'last_login': user.last_login.isoformat() if user.last_login else None, } def _product_to_dict(product, user): """ Serialize a Product object, including can_edit / can_delete flags based on whether the user created the product (via LogEntry). """ is_owner = _user_owns_object(user, 'product', product.id) return { 'id': product.id, 'description_text': product.description_text, 'ncm': product.ncm, 'cfop': product.cfop, 'ean': product.ean or '', 'unidademedida': product.unidademedida or '', 'can_edit': is_owner, 'can_delete': is_owner, } # --------------------------------------------------------------------------- # Ownership helpers (LogEntry-based, same pattern as admin.py) # --------------------------------------------------------------------------- def _get_family_user_ids(user): """Return IDs of all members of the user's family group, or just [user.id].""" family_group = FamilyGroup.objects.filter(group__user=user).select_related('group').first() if not family_group: return [user.id] return list(family_group.group.user_set.values_list('id', flat=True)) def _user_owns_object(user, model_name, object_id): """Check if the user (or a family member) created a given object (via LogEntry ADDITION).""" if user.is_superuser: return True family_ids = _get_family_user_ids(user) ct = ContentType.objects.get(app_label='notas', model=model_name) return LogEntry.objects.filter( user_id__in=family_ids, action_flag=ADDITION, content_type_id=ct.id, object_id=str(object_id), ).exists() def _get_user_order_ids(user): """Return a list of Order IDs created by the user or family members (via LogEntry).""" if user.is_superuser: return list(Order.objects.values_list('id', flat=True)) family_ids = _get_family_user_ids(user) ct = ContentType.objects.get(app_label='notas', model='order') object_ids = LogEntry.objects.filter( user_id__in=family_ids, action_flag=ADDITION, content_type_id=ct.id, ).values_list('object_id', flat=True) return [int(oid) for oid in object_ids] def _get_order_importer(order_id): """ Resolve the user who imported/created an Order using django_admin_log. Uses the first ADDITION LogEntry for the Order object as the importer source. Returns a dict with basic user information or None if not found. """ ct = ContentType.objects.get(app_label='notas', model='order') importer_log = ( LogEntry.objects .select_related('user') .filter( content_type_id=ct.id, object_id=str(order_id), action_flag=ADDITION, ) .order_by('action_time', 'id') .first() ) if not importer_log or not importer_log.user_id: return None importer = importer_log.user return { 'id': importer.id, 'name': importer.get_full_name() or importer.username, 'email': importer.email or '', } def _get_user_product_ids(user): """ Return product IDs that belong to the user or family members: 1. Products from the family's orders (bought). 2. Products the family created directly (via LogEntry on Product). """ # Products from orders order_ids = _get_user_order_ids(user) from_orders = set( OrderProduct.objects .filter(order_id__in=order_ids) .values_list('product_id', flat=True) ) # Products the user or family created (registered via app or admin) family_ids = _get_family_user_ids(user) ct_product = ContentType.objects.get(app_label='notas', model='product') created_ids = set( int(oid) for oid in LogEntry.objects.filter( user_id__in=family_ids, action_flag=ADDITION, content_type_id=ct_product.id, ).values_list('object_id', flat=True) ) return list(from_orders | created_ids) # --------------------------------------------------------------------------- # Auth views # --------------------------------------------------------------------------- @csrf_exempt @require_POST @require_api_key def google_auth(request): """ Authenticate a user via Google ID token. Expects JSON body: { "id_token": "" } - Verifies the token with Google. - If the user (by email) already exists in auth_user, updates last_login. - If the user doesn't exist, creates a new record in auth_user. - Returns a JWT token and user info. """ try: body = json.loads(request.body) except (json.JSONDecodeError, ValueError): return JsonResponse({'error': 'JSON inválido.'}, status=400) google_token = body.get('id_token', '').strip() if not google_token: return JsonResponse({'error': 'id_token é obrigatório.'}, status=400) # Verify the Google ID token try: idinfo = id_token.verify_oauth2_token( google_token, google_requests.Request(), settings.GOOGLE_CLIENT_ID, ) except ValueError as e: return JsonResponse( {'error': f'Token do Google inválido: {str(e)}'}, status=401, ) email = idinfo.get('email', '').lower() if not email or not idinfo.get('email_verified', False): return JsonResponse( {'error': 'E-mail não verificado pelo Google.'}, status=401, ) first_name = idinfo.get('given_name', '') last_name = idinfo.get('family_name', '') picture = idinfo.get('picture', '') # Look up or create the user try: user = User.objects.get(email=email) # Update last_login user.last_login = timezone.now() # Update name/picture if changed if first_name: user.first_name = first_name if last_name: user.last_name = last_name user.save(update_fields=['last_login', 'first_name', 'last_name']) created = False except User.DoesNotExist: # Create new user username = email.split('@')[0] # Ensure unique username base_username = username counter = 1 while User.objects.filter(username=username).exists(): username = f'{base_username}{counter}' counter += 1 user = User.objects.create_user( username=username, email=email, first_name=first_name, last_name=last_name, password=None, # No password — Google-only auth ) user.last_login = timezone.now() user.save(update_fields=['last_login']) created = True # Auto-accept pending family invites only for newly registered users if created: auto_invites = FamilyInvite.objects.filter( invited_email__iexact=email, status='pending', ).select_related('family__group') for invite in auto_invites: if not FamilyGroup.objects.filter(group__user=user).exists(): invite.status = 'accepted' invite.save(update_fields=['status']) invite.family.group.user_set.add(user) # Generate JWT for the app token = _generate_jwt(user) return JsonResponse({ 'token': token, 'user': { **_user_to_dict(user), 'picture': picture, }, 'created': created, }) @csrf_exempt @require_POST @require_api_key def apple_auth(request): """ Authenticate a user via Sign in with Apple identity token. Expects JSON body: { "identity_token": "", "full_name": "Fulano de Tal" # optional, only sent on first login } The identity_token is a JWT signed by Apple (RS256). We verify: - Signature against Apple's public JWKs. - issuer == https://appleid.apple.com - audience == settings.APPLE_BUNDLE_ID - expiration Apple only includes the user's email on the first sign-in. On subsequent sign-ins the email may be missing. We always have `sub` (a stable Apple user id), but since we currently key Django users by email we require an email on the first request. Apps that use "Hide my email" will receive a proxy address (@privaterelay.appleid.com) which behaves like any other email for our purposes. """ try: body = json.loads(request.body) except (json.JSONDecodeError, ValueError): return JsonResponse({'error': 'JSON inválido.'}, status=400) identity_token = body.get('identity_token', '').strip() full_name = (body.get('full_name') or '').strip() if not identity_token: return JsonResponse({'error': 'identity_token é obrigatório.'}, status=400) try: signing_key = _get_apple_jwks_client().get_signing_key_from_jwt(identity_token) payload = jwt.decode( identity_token, signing_key.key, algorithms=['RS256'], audience=settings.APPLE_BUNDLE_ID, issuer='https://appleid.apple.com', ) except jwt.ExpiredSignatureError: return JsonResponse({'error': 'Token Apple expirado.'}, status=401) except jwt.InvalidAudienceError: return JsonResponse({'error': 'Token Apple não é deste app.'}, status=401) except jwt.InvalidIssuerError: return JsonResponse({'error': 'Emissor do token Apple inválido.'}, status=401) except (jwt.InvalidTokenError, jwt.PyJWKClientError) as e: return JsonResponse( {'error': f'Token Apple inválido: {str(e)}'}, status=401, ) apple_sub = payload.get('sub', '') email = (payload.get('email') or '').lower() email_verified = str(payload.get('email_verified', 'false')).lower() == 'true' if not email: return JsonResponse( {'error': 'E-mail não disponível no token Apple. Revogue o acesso nas ' 'configurações da sua Apple ID e tente novamente.'}, status=401, ) # Apple either marks email as verified or sets it to the proxy domain. if not email_verified and not email.endswith('@privaterelay.appleid.com'): return JsonResponse( {'error': 'E-mail não verificado pela Apple.'}, status=401, ) first_name, last_name = '', '' if full_name: parts = full_name.split(' ', 1) first_name = parts[0] last_name = parts[1] if len(parts) > 1 else '' try: user = User.objects.get(email=email) user.last_login = timezone.now() update_fields = ['last_login'] if first_name and not user.first_name: user.first_name = first_name update_fields.append('first_name') if last_name and not user.last_name: user.last_name = last_name update_fields.append('last_name') user.save(update_fields=update_fields) created = False except User.DoesNotExist: base_username = (email.split('@')[0] or f'apple_{apple_sub[:8]}') username = base_username counter = 1 while User.objects.filter(username=username).exists(): username = f'{base_username}{counter}' counter += 1 user = User.objects.create_user( username=username, email=email, first_name=first_name, last_name=last_name, password=None, ) user.last_login = timezone.now() user.save(update_fields=['last_login']) created = True # Auto-accept pending family invites for the new account. auto_invites = FamilyInvite.objects.filter( invited_email__iexact=email, status='pending', ).select_related('family__group') for invite in auto_invites: if not FamilyGroup.objects.filter(group__user=user).exists(): invite.status = 'accepted' invite.save(update_fields=['status']) invite.family.group.user_set.add(user) token = _generate_jwt(user) return JsonResponse({ 'token': token, 'user': { **_user_to_dict(user), 'picture': '', }, 'created': created, }) @csrf_exempt @require_GET @require_api_key def me(request): """ Return the current authenticated user's info. Expects header: X-Auth-Token: """ token = request.headers.get('X-Auth-Token', '').strip() if not token: return JsonResponse({'error': 'Token não fornecido.'}, status=401) payload = _decode_jwt(token) if payload is None: return JsonResponse({'error': 'Token inválido ou expirado.'}, status=401) try: user = User.objects.get(id=payload['user_id']) except User.DoesNotExist: return JsonResponse({'error': 'Usuário não encontrado.'}, status=404) return JsonResponse({'user': _user_to_dict(user)}) @csrf_exempt @require_auth def delete_account(request): """ DELETE /api/auth/account/ — Permanently delete the user's account and all associated data (orders, order products, products, vendors, log entries). Uses django_admin_log to find everything the user created. """ if request.method != 'DELETE': return JsonResponse({'error': 'Método não permitido.'}, status=405) user = request.api_user # 1. Find all object IDs the user created, grouped by content type user_logs = LogEntry.objects.filter(user_id=user.id, action_flag=ADDITION) ct_order = ContentType.objects.get(app_label='notas', model='order') ct_product = ContentType.objects.get(app_label='notas', model='product') ct_vendor = ContentType.objects.get(app_label='notas', model='vendor') order_ids = [ int(e.object_id) for e in user_logs.filter(content_type_id=ct_order.id) ] product_ids = [ int(e.object_id) for e in user_logs.filter(content_type_id=ct_product.id) ] vendor_ids = [ int(e.object_id) for e in user_logs.filter(content_type_id=ct_vendor.id) ] # 2. Delete in dependency order # OrderProduct rows for the user's orders OrderProduct.objects.filter(order_id__in=order_ids).delete() # Orders Order.objects.filter(id__in=order_ids).delete() # Products the user created Product.objects.filter(id__in=product_ids).delete() # Vendors the user created Vendor.objects.filter(id__in=vendor_ids).delete() # 3. Delete notifications and user profile Notification.objects.filter(user=user).delete() UserProfile.objects.filter(user=user).delete() # 4. Delete all LogEntry records for this user LogEntry.objects.filter(user_id=user.id).delete() # 5. Delete the user account itself user.delete() return JsonResponse({'detail': 'Conta e todos os dados foram excluídos com sucesso.'}) # --------------------------------------------------------------------------- # Location / Radius views # --------------------------------------------------------------------------- @csrf_exempt @require_POST @require_auth def update_location(request): """ POST /api/auth/location/ — Update user's location and optionally push token. Expects JSON body: { "latitude": float, "longitude": float, "expo_push_token"?: string } Creates or updates the UserProfile. """ try: body = json.loads(request.body) except (json.JSONDecodeError, ValueError): return JsonResponse({'error': 'JSON inválido.'}, status=400) lat = body.get('latitude') lng = body.get('longitude') push_token = body.get('expo_push_token', '').strip() if lat is None or lng is None: return JsonResponse({'error': 'latitude e longitude são obrigatórios.'}, status=400) try: lat = float(lat) lng = float(lng) except (ValueError, TypeError): return JsonResponse({'error': 'latitude e longitude devem ser números.'}, status=400) profile, _ = UserProfile.objects.get_or_create(user=request.api_user) profile.latitude = lat profile.longitude = lng if push_token: profile.expo_push_token = push_token profile.save() return JsonResponse(_profile_to_dict(profile)) def _profile_to_dict(profile): """Serialize UserProfile for API responses.""" return { 'latitude': profile.latitude, 'longitude': profile.longitude, 'radius_km': profile.radius_km, 'notifications_enabled': profile.notifications_enabled, 'expo_push_token': profile.expo_push_token, } @csrf_exempt @require_auth def notification_settings(request): """ GET /api/auth/notification-settings/ — Fetch current notification preferences. PATCH /api/auth/notification-settings/ — Update notifications_enabled and/or radius_km. PATCH expects JSON body: { "notifications_enabled"?: bool, "radius_km"?: int } """ profile, _ = UserProfile.objects.get_or_create(user=request.api_user) if request.method == 'GET': return JsonResponse(_profile_to_dict(profile)) if request.method == 'PATCH': try: body = json.loads(request.body) except (json.JSONDecodeError, ValueError): return JsonResponse({'error': 'JSON inválido.'}, status=400) update_fields = [] if 'notifications_enabled' in body: profile.notifications_enabled = bool(body['notifications_enabled']) update_fields.append('notifications_enabled') if 'radius_km' in body: try: radius = int(body['radius_km']) except (ValueError, TypeError): return JsonResponse({'error': 'radius_km deve ser um número inteiro.'}, status=400) if radius < 1 or radius > 25: return JsonResponse({'error': 'radius_km deve estar entre 1 e 25.'}, status=400) profile.radius_km = radius update_fields.append('radius_km') if update_fields: profile.save(update_fields=update_fields) return JsonResponse(_profile_to_dict(profile)) return JsonResponse({'error': 'Método não permitido.'}, status=405) @csrf_exempt @require_auth def goals_view(request): """ GET /api/auth/goals/ — Return vendor and product spending goals. PATCH /api/auth/goals/ — Update goals. Accepts partial JSON: { "vendors_goal"?: [...], "products_goal"?: [...] } """ profile, _ = UserProfile.objects.get_or_create(user=request.api_user) def _parse_goals(raw): try: data = json.loads(raw) if isinstance(raw, str) else raw return data if isinstance(data, list) else [] except (json.JSONDecodeError, ValueError, TypeError): return [] if request.method == 'GET': return JsonResponse({ 'vendors_goal': _parse_goals(profile.vendors_goal), 'products_goal': _parse_goals(profile.products_goal), }) if request.method == 'PATCH': try: body = json.loads(request.body) except (json.JSONDecodeError, ValueError): return JsonResponse({'error': 'JSON inválido.'}, status=400) update_fields = [] if 'vendors_goal' in body: goals = body['vendors_goal'] if not isinstance(goals, list): return JsonResponse({'error': 'vendors_goal deve ser uma lista.'}, status=400) profile.vendors_goal = json.dumps(goals) update_fields.append('vendors_goal') if 'products_goal' in body: goals = body['products_goal'] if not isinstance(goals, list): return JsonResponse({'error': 'products_goal deve ser uma lista.'}, status=400) profile.products_goal = json.dumps(goals) update_fields.append('products_goal') if update_fields: profile.save(update_fields=update_fields) return JsonResponse({ 'vendors_goal': _parse_goals(profile.vendors_goal), 'products_goal': _parse_goals(profile.products_goal), }) return JsonResponse({'error': 'Método não permitido.'}, status=405) @csrf_exempt @require_http_methods(['PATCH']) @require_auth def update_radius(request): """ PATCH /api/auth/radius/ — Update notification radius (1-25 km). Expects JSON body: { "radius_km": int } """ try: body = json.loads(request.body) except (json.JSONDecodeError, ValueError): return JsonResponse({'error': 'JSON inválido.'}, status=400) radius = body.get('radius_km') if radius is None: return JsonResponse({'error': 'radius_km é obrigatório.'}, status=400) try: radius = int(radius) except (ValueError, TypeError): return JsonResponse({'error': 'radius_km deve ser um número inteiro.'}, status=400) if radius < 1 or radius > 25: return JsonResponse({'error': 'radius_km deve estar entre 1 e 25.'}, status=400) profile, _ = UserProfile.objects.get_or_create(user=request.api_user) profile.radius_km = radius profile.save(update_fields=['radius_km']) return JsonResponse(_profile_to_dict(profile)) # --------------------------------------------------------------------------- # Product views # --------------------------------------------------------------------------- @csrf_exempt @require_auth def product_list_create(request): """ GET /api/produtos/ — List products from the user's orders. POST /api/produtos/ — Create a new product (+ LogEntry for ownership). """ if request.method == 'GET': return _product_list(request) elif request.method == 'POST': return _product_create(request) else: return JsonResponse({'error': 'Método não permitido.'}, status=405) @csrf_exempt @require_auth def product_detail(request, product_id): """ GET /api/produtos/:id/ — Product detail with can_edit/can_delete. PATCH /api/produtos/:id/ — Update description_text (owner only). DELETE /api/produtos/:id/ — Delete product (owner only). """ try: product = Product.objects.get(id=product_id) except Product.DoesNotExist: return JsonResponse({'error': 'Produto não encontrado.'}, status=404) if request.method == 'GET': return JsonResponse(_product_to_dict(product, request.api_user)) elif request.method == 'PATCH': return _product_update(request, product) elif request.method == 'DELETE': return _product_delete(request, product) else: return JsonResponse({'error': 'Método não permitido.'}, status=405) def _parse_period(period_str): """Parse a period string and return (start_date, end_date) or (None, None) for 'all'.""" if not period_str or period_str == 'all': return None, None now = timezone.now() if period_str == 'month': start = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0) return start, None if period_str == '30d': start = now - datetime.timedelta(days=30) return start, None if period_str == '6m': start = now - datetime.timedelta(days=183) return start, None # Year range: "2024-2026" m = re.match(r'^(\d{4})-(\d{4})$', period_str) if m: y1, y2 = int(m.group(1)), int(m.group(2)) start = datetime.datetime(y1, 1, 1, tzinfo=datetime.timezone.utc) end = datetime.datetime(y2, 12, 31, 23, 59, 59, tzinfo=datetime.timezone.utc) return start, end # Single year: "2026" m = re.match(r'^(\d{4})$', period_str) if m: y = int(m.group(1)) start = datetime.datetime(y, 1, 1, tzinfo=datetime.timezone.utc) end = datetime.datetime(y, 12, 31, 23, 59, 59, tzinfo=datetime.timezone.utc) return start, end return None, None def _filter_order_ids_by_period(order_ids, period_str): """Filter order IDs by the given period string.""" start, end = _parse_period(period_str) if start is None and end is None: return order_ids qs = Order.objects.filter(id__in=order_ids) if start: qs = qs.filter(sale_date__gte=start) if end: qs = qs.filter(sale_date__lte=end) return list(qs.values_list('id', flat=True)) def _product_list(request): """Return all products the user has purchased (via their orders), with price stats.""" user = request.api_user product_ids = _get_user_product_ids(user) order_ids = _get_user_order_ids(user) period = request.GET.get('period', 'all') order_ids = _filter_order_ids_by_period(order_ids, period) # Re-derive product_ids from filtered orders so products without purchases # in the selected period are excluded if period and period != 'all': product_ids = list( OrderProduct.objects .filter(order_id__in=order_ids) .values_list('product_id', flat=True) .distinct() ) # Pre-compute price stats for all products in a single query stats_qs = ( OrderProduct.objects .filter(product_id__in=product_ids, order_id__in=order_ids) .values('product_id') .annotate( min_price=Min('product_un_price'), max_price=Max('product_un_price'), avg_price=Avg('product_un_price'), first_purchase=Min('order__sale_date'), total_quantity=Sum('product_quantity'), total_spent=Sum('product_total_price'), ) ) stats_map = { row['product_id']: row for row in stats_qs } products = Product.objects.filter(id__in=product_ids).order_by('description_text') data = [] for p in products: d = _product_to_dict(p, user) st = stats_map.get(p.id) if st: d['min_price'] = round(st['min_price'], 2) if st['min_price'] is not None else None d['max_price'] = round(st['max_price'], 2) if st['max_price'] is not None else None d['avg_price'] = round(st['avg_price'], 2) if st['avg_price'] is not None else None d['first_purchase'] = st['first_purchase'].isoformat() if st['first_purchase'] else None d['total_quantity'] = round(st['total_quantity'], 3) if st['total_quantity'] is not None else None d['total_spent'] = round(st['total_spent'], 2) if st['total_spent'] is not None else None else: d['min_price'] = None d['max_price'] = None d['avg_price'] = None d['first_purchase'] = None d['total_quantity'] = None d['total_spent'] = None data.append(d) return JsonResponse(data, safe=False) def _product_create(request): """Create a new product and record a LogEntry so admin ownership is tracked.""" try: body = json.loads(request.body) except (json.JSONDecodeError, ValueError): return JsonResponse({'error': 'JSON inválido.'}, status=400) description = body.get('description_text', '').strip() if not description: return JsonResponse( {'error': 'description_text é obrigatório.'}, status=400, ) ean = body.get('ean', '').strip() if not ean: return JsonResponse({'error': 'ean é obrigatório.'}, status=400) ncm = body.get('ncm', 0) if not ncm: return JsonResponse({'error': 'ncm é obrigatório.'}, status=400) product = Product.objects.create( description_text=description, ncm=int(ncm), cfop=int(body.get('cfop', 0)), ean=ean, unidademedida=body.get('unidademedida', 'Un'), ) # Record LogEntry (ADDITION) so admin ownership tracking stays consistent ct = ContentType.objects.get_for_model(Product) LogEntry.objects.create( user_id=request.api_user.id, content_type_id=ct.id, object_id=str(product.id), object_repr=str(product), action_flag=ADDITION, ) return JsonResponse( _product_to_dict(product, request.api_user), status=201, ) def _product_update(request, product): """Update only description_text. Requires ownership.""" user = request.api_user if not _user_owns_object(user, 'product', product.id): return JsonResponse( {'error': 'Você não tem permissão para editar este produto.'}, status=403, ) try: body = json.loads(request.body) except (json.JSONDecodeError, ValueError): return JsonResponse({'error': 'JSON inválido.'}, status=400) description = body.get('description_text', '').strip() if not description: return JsonResponse( {'error': 'description_text é obrigatório.'}, status=400, ) product.description_text = description product.save(update_fields=['description_text']) return JsonResponse(_product_to_dict(product, user)) def _product_delete(request, product): """Delete the product. Requires ownership.""" user = request.api_user if not _user_owns_object(user, 'product', product.id): return JsonResponse( {'error': 'Você não tem permissão para excluir este produto.'}, status=403, ) product.delete() return JsonResponse({'detail': 'Produto excluído com sucesso.'}, status=200) # Deterministic color palette for vendor chart lines _VENDOR_COLORS = [ '#2E7D32', '#1976D2', '#D32F2F', '#F9A825', '#7B1FA2', '#00838F', '#E64A19', '#5D4037', '#455A64', '#C2185B', '#00695C', '#283593', '#EF6C00', '#6A1B9A', '#1565C0', ] @csrf_exempt @require_GET @require_auth def product_history(request, product_id): """ GET /api/produtos//historico/ — Price history for a product. Returns stats (min, max, avg, count), a chronological list of purchases with vendor info, and a deduplicated vendor list with assigned colors. Scoped to the authenticated user's orders only. """ try: product = Product.objects.get(id=product_id) except Product.DoesNotExist: return JsonResponse({'error': 'Produto não encontrado.'}, status=404) user = request.api_user order_ids = _get_user_order_ids(user) entries = ( OrderProduct.objects .filter(product_id=product_id, order_id__in=order_ids) .select_related('order', 'order__vendor') .order_by('order__sale_date') ) # Build history list and collect vendor ids history = [] vendor_map = {} # id -> name for op in entries: vendor = op.order.vendor vendor_name = vendor.nomefantasia_text or vendor.nomerazaosocial_text or vendor.cpfcnpj_text vendor_map[vendor.id] = vendor_name history.append({ 'date': op.order.sale_date.isoformat(), 'unit_price': round(op.product_un_price, 2), 'quantity': round(op.product_quantity, 3), 'total_price': round(op.product_total_price, 2), 'vendor_id': vendor.id, 'vendor_name': vendor_name, }) # Compute stats from the same queryset agg = ( OrderProduct.objects .filter(product_id=product_id, order_id__in=order_ids) .aggregate( min_price=Min('product_un_price'), max_price=Max('product_un_price'), avg_price=Avg('product_un_price'), total_purchases=Count('id'), ) ) stats = { 'min_price': round(agg['min_price'], 2) if agg['min_price'] is not None else None, 'max_price': round(agg['max_price'], 2) if agg['max_price'] is not None else None, 'avg_price': round(agg['avg_price'], 2) if agg['avg_price'] is not None else None, 'total_purchases': agg['total_purchases'], } # Build vendor list with deterministic colors vendors = [] for idx, (vid, vname) in enumerate(sorted(vendor_map.items())): vendors.append({ 'id': vid, 'name': vname, 'color': _VENDOR_COLORS[idx % len(_VENDOR_COLORS)], }) return JsonResponse({ 'stats': stats, 'history': history, 'vendors': vendors, }) # --------------------------------------------------------------------------- # Haversine distance # --------------------------------------------------------------------------- def _haversine_km(lat1, lon1, lat2, lon2): """Compute the Haversine distance in km between two (lat, lon) points.""" R = 6371.0 # Earth radius in km dlat = math.radians(lat2 - lat1) dlon = math.radians(lon2 - lon1) a = (math.sin(dlat / 2) ** 2 + math.cos(math.radians(lat1)) * math.cos(math.radians(lat2)) * math.sin(dlon / 2) ** 2) return R * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a)) # --------------------------------------------------------------------------- # Notification trigger # --------------------------------------------------------------------------- def _check_and_notify(current_user, order, product_objs): """ After a new order is imported, check if any product has a lower price than what other users have previously paid. If so, and the vendor is within the user's radius, send a push notification and create a Notification record. This runs in a background thread to avoid blocking the import response. """ try: vendor = order.vendor if not vendor.latitude or not vendor.longitude: return # Can't compute distance without vendor coords # Content types for LogEntry queries ct_order = ContentType.objects.get(app_label='notas', model='order') for product, item in product_objs: new_price = float(item.get('valor_unitario', '0').replace(',', '.') if isinstance(item.get('valor_unitario'), str) else item.get('valor_unitario', 0)) if new_price <= 0: continue # Find all OTHER users who have bought this product # (via LogEntry on Order -> OrderProduct with this product) other_user_ids = set( LogEntry.objects.filter( content_type_id=ct_order.id, action_flag=ADDITION, ).exclude( user_id=current_user.id, ).values_list('user_id', flat=True).distinct() ) for other_user_id in other_user_ids: try: other_profile = UserProfile.objects.get(user_id=other_user_id) except UserProfile.DoesNotExist: continue if not other_profile.notifications_enabled: continue if not other_profile.latitude or not other_profile.longitude: continue if not other_profile.expo_push_token: continue # Check distance distance = _haversine_km( other_profile.latitude, other_profile.longitude, vendor.latitude, vendor.longitude, ) if distance > other_profile.radius_km: continue # Get the other user's order IDs other_order_ids = [ int(oid) for oid in LogEntry.objects.filter( user_id=other_user_id, content_type_id=ct_order.id, action_flag=ADDITION, ).values_list('object_id', flat=True) ] # Check if this user has ever bought this product best = OrderProduct.objects.filter( product_id=product.id, order_id__in=other_order_ids, ).aggregate(min_price=Min('product_un_price')) user_best_price = best.get('min_price') if user_best_price is None: continue # User never bought this product if new_price >= user_best_price: continue # Not a better price # Create notification vendor_name = vendor.nomefantasia_text or vendor.nomerazaosocial_text or vendor.cpfcnpj_text title = 'Preço mais baixo encontrado!' body = ( f'Alguém na sua região adquiriu {product.description_text} ' f'por R$ {new_price:.2f}. Clique para saber onde e quando.' ) notif = Notification.objects.create( user_id=other_user_id, title=title, body=body, product=product, vendor=vendor, price=new_price, ) # Send push notification send_push_notification( other_profile.expo_push_token, title, body, data={ 'notification_id': notif.id, 'product_id': product.id, 'vendor_id': vendor.id, }, ) except Exception as e: print(f'[Notify] Error in _check_and_notify: {e}') def _check_goals(user, order, product_objs): """ After a new order is imported, check whether the current user's spending goals (vendor / product) have reached 80 % or 100 %. If so, create a Notification record and send a push notification. Runs in a background thread to avoid blocking the import response. """ try: profile = UserProfile.objects.filter(user=user).first() if not profile or not profile.expo_push_token or not profile.notifications_enabled: return def _parse_goals(raw): try: data = json.loads(raw) if isinstance(raw, str) else raw return data if isinstance(data, list) else [] except Exception: return [] PERIOD_DELTAS = { '30d': datetime.timedelta(days=30), '60d': datetime.timedelta(days=60), '90d': datetime.timedelta(days=90), '6m': datetime.timedelta(days=183), '1y': datetime.timedelta(days=365), } def _cutoff(period_str): delta = PERIOD_DELTAS.get(period_str) if delta: return (timezone.now() - delta).date() return None def _already_notified(user_id, title_prefix, vendor=None, product=None, since=None): qs = Notification.objects.filter(user_id=user_id, title__startswith=title_prefix) if vendor: qs = qs.filter(vendor=vendor) if product: qs = qs.filter(product=product) if since: qs = qs.filter(created_at__date__gte=since) return qs.exists() order_ids = _get_user_order_ids(user) # ---- Vendor goals ---- vendors_goal = _parse_goals(profile.vendors_goal) for vg in vendors_goal: vid = vg.get('vendor_id') goal_value = vg.get('goal', 0) period = vg.get('period', '30d') if not vid or not goal_value: continue if vid != order.vendor_id: continue cutoff = _cutoff(period) if not cutoff: continue total_qs = Order.objects.filter(id__in=order_ids, vendor_id=vid, sale_date__gte=cutoff) total_spent = total_qs.aggregate(s=Sum('total_paid'))['s'] or 0 pct = (total_spent / goal_value) * 100 if goal_value else 0 vendor_obj = order.vendor vendor_name = vendor_obj.nomefantasia_text or vendor_obj.nomerazaosocial_text or vendor_obj.cpfcnpj_text if pct >= 100: prefix = 'Meta estourada' if _already_notified(user.id, prefix, vendor=vendor_obj, since=cutoff): continue title = f'{prefix}!' body = ( f'Você gastou R$ {total_spent:.2f} em {vendor_name}, ' f'acima da meta de R$ {goal_value:.2f}.' ) elif pct >= 80: prefix = 'Atenção: meta' if _already_notified(user.id, prefix, vendor=vendor_obj, since=cutoff): continue title = f'{prefix} próxima!' body = ( f'Você já gastou {pct:.0f}% da sua meta de R$ {goal_value:.2f} ' f'em {vendor_name}.' ) else: continue notif = Notification.objects.create( user=user, title=title, body=body, vendor=vendor_obj, price=total_spent, ) send_push_notification( profile.expo_push_token, title, body, data={'notification_id': notif.id, 'vendor_id': vid}, ) # ---- Product goals ---- products_goal = _parse_goals(profile.products_goal) product_ids_in_order = {p.id for p, _ in product_objs} for pg in products_goal: pid = pg.get('product_id') goal_value = pg.get('goal', 0) period = pg.get('period', '30d') if not pid or not goal_value: continue if pid not in product_ids_in_order: continue cutoff = _cutoff(period) if not cutoff: continue total_spent = ( OrderProduct.objects.filter( order_id__in=order_ids, product_id=pid, order__sale_date__gte=cutoff, ).aggregate(s=Sum('product_total_price'))['s'] or 0 ) pct = (total_spent / goal_value) * 100 if goal_value else 0 product_obj = Product.objects.filter(id=pid).first() if not product_obj: continue product_name = product_obj.description_text if pct >= 100: prefix = 'Meta estourada' if _already_notified(user.id, prefix, product=product_obj, since=cutoff): continue title = f'{prefix}!' body = ( f'Você gastou R$ {total_spent:.2f} em {product_name}, ' f'acima da meta de R$ {goal_value:.2f}.' ) elif pct >= 80: prefix = 'Atenção: meta' if _already_notified(user.id, prefix, product=product_obj, since=cutoff): continue title = f'{prefix} próxima!' body = ( f'Você já gastou {pct:.0f}% da sua meta de R$ {goal_value:.2f} ' f'em {product_name}.' ) else: continue notif = Notification.objects.create( user=user, title=title, body=body, product=product_obj, price=total_spent, ) send_push_notification( profile.expo_push_token, title, body, data={'notification_id': notif.id, 'product_id': pid}, ) except Exception as e: print(f'[Goals] Error in _check_goals: {e}') # --------------------------------------------------------------------------- # Notification API views # --------------------------------------------------------------------------- @csrf_exempt @require_GET @require_auth def notification_list(request): """ GET /api/notificacoes/ — List user's notifications (newest first). Optional query params: ?page=1&page_size=20 """ page = int(request.GET.get('page', 1)) page_size = int(request.GET.get('page_size', 20)) offset = (page - 1) * page_size notifications = ( Notification.objects .filter(user=request.api_user) .select_related('product', 'vendor') .order_by('-created_at') ) total = notifications.count() items = notifications[offset:offset + page_size] data = [] for n in items: vendor_name = '' if n.vendor: vendor_name = n.vendor.nomefantasia_text or n.vendor.nomerazaosocial_text or n.vendor.cpfcnpj_text or '' data.append({ 'id': n.id, 'title': n.title, 'body': n.body, 'product_id': n.product_id, 'product_name': n.product.description_text if n.product else '', 'vendor_id': n.vendor_id, 'vendor_name': vendor_name, 'price': n.price, 'is_read': n.is_read, 'created_at': n.created_at.isoformat(), }) return JsonResponse({ 'results': data, 'total': total, 'page': page, 'page_size': page_size, }) @csrf_exempt @require_GET @require_auth def notification_detail(request, notification_id): """ GET /api/notificacoes// — Return a single notification with full vendor details (address, latitude, longitude) for the deal screen. """ try: notif = Notification.objects.select_related('product', 'vendor').get( id=notification_id, user=request.api_user ) except Notification.DoesNotExist: return JsonResponse({'error': 'Notificação não encontrada.'}, status=404) # Mark as read on access if not notif.is_read: notif.is_read = True notif.save(update_fields=['is_read']) vendor_name = '' vendor_address = '' vendor_latitude = None vendor_longitude = None if notif.vendor: vendor_name = ( notif.vendor.nomefantasia_text or notif.vendor.nomerazaosocial_text or notif.vendor.cpfcnpj_text or '' ) # Build address string from vendor fields parts = [] if notif.vendor.endereco_text: parts.append(notif.vendor.endereco_text) if notif.vendor.bairro_text: parts.append(notif.vendor.bairro_text) city_state = [] if notif.vendor.municipio_text: city_state.append(notif.vendor.municipio_text) if notif.vendor.uf_text: city_state.append(notif.vendor.uf_text) if city_state: parts.append(' - '.join(city_state)) if notif.vendor.cep_text: parts.append(f'CEP {notif.vendor.cep_text}') vendor_address = ', '.join(parts) vendor_latitude = notif.vendor.latitude vendor_longitude = notif.vendor.longitude return JsonResponse({ 'id': notif.id, 'title': notif.title, 'body': notif.body, 'product_id': notif.product_id, 'product_name': notif.product.description_text if notif.product else '', 'vendor_id': notif.vendor_id, 'vendor_name': vendor_name, 'vendor_address': vendor_address, 'vendor_latitude': vendor_latitude, 'vendor_longitude': vendor_longitude, 'price': notif.price, 'is_read': notif.is_read, 'created_at': notif.created_at.isoformat(), }) @csrf_exempt @require_GET @require_auth def notification_unread_count(request): """GET /api/notificacoes/unread-count/ — Returns { count: N }.""" count = Notification.objects.filter( user=request.api_user, is_read=False, ).count() return JsonResponse({'count': count}) @csrf_exempt @require_http_methods(['PATCH']) @require_auth def notification_mark_read(request, notification_id): """PATCH /api/notificacoes//read/ — Mark a single notification as read.""" try: notif = Notification.objects.get(id=notification_id, user=request.api_user) except Notification.DoesNotExist: return JsonResponse({'error': 'Notificação não encontrada.'}, status=404) notif.is_read = True notif.save(update_fields=['is_read']) return JsonResponse({'id': notif.id, 'is_read': True}) @csrf_exempt @require_POST @require_auth def notification_read_all(request): """POST /api/notificacoes/read-all/ — Mark all user's notifications as read.""" updated = Notification.objects.filter( user=request.api_user, is_read=False, ).update(is_read=True) return JsonResponse({'updated': updated}) @csrf_exempt @require_http_methods(['DELETE']) @require_auth def notification_delete(request, notification_id): """DELETE /api/notificacoes//delete/ — Delete a single notification.""" deleted, _ = Notification.objects.filter( id=notification_id, user=request.api_user ).delete() if not deleted: return JsonResponse({'error': 'Notificação não encontrada.'}, status=404) return JsonResponse({'deleted': True}) @csrf_exempt @require_http_methods(['DELETE']) @require_auth def notification_delete_all(request): """DELETE /api/notificacoes/delete-all/ — Delete all user's notifications.""" deleted, _ = Notification.objects.filter(user=request.api_user).delete() return JsonResponse({'deleted': deleted}) # --------------------------------------------------------------------------- # Vendor (Estabelecimento) views # --------------------------------------------------------------------------- @csrf_exempt @require_auth def vendor_list(request): """ GET /api/estabelecimentos/ — List vendors where the user has made purchases, including total spent at each. """ if request.method != 'GET': return JsonResponse({'error': 'Method not allowed'}, status=405) user = request.api_user order_ids = _get_user_order_ids(user) period = request.GET.get('period', 'all') order_ids = _filter_order_ids_by_period(order_ids, period) if not order_ids: return JsonResponse([], safe=False) # Get vendors with aggregated totals from the user's orders vendors = ( Order.objects.filter(id__in=order_ids) .values('vendor_id') .annotate( total_spent=Sum('total_paid'), order_count=Count('id'), last_purchase=Max('sale_date'), ) .order_by('-last_purchase') ) vendor_ids = [v['vendor_id'] for v in vendors] vendor_objs = {v.id: v for v in Vendor.objects.filter(id__in=vendor_ids)} result = [] for row in vendors: v = vendor_objs.get(row['vendor_id']) if not v: continue result.append({ 'id': v.id, 'nome_fantasia': v.nomefantasia_text or '', 'razao_social': v.nomerazaosocial_text or '', 'cnpj': v.cpfcnpj_text or '', 'bairro': v.bairro_text or '', 'endereco': v.endereco_text or '', 'municipio': v.municipio_text or '', 'uf': v.uf_text or '', 'latitude': v.latitude, 'longitude': v.longitude, 'total_spent': round(row['total_spent'] or 0, 2), 'order_count': row['order_count'], 'last_purchase': row['last_purchase'].isoformat() if row['last_purchase'] else None, }) return JsonResponse(result, safe=False) @csrf_exempt @require_auth def vendor_detail(request, vendor_id): """ GET /api/estabelecimentos/:id/ — Vendor detail with can_edit flag. PATCH /api/estabelecimentos/:id/ — Update nomefantasia (owner only). """ try: vendor = Vendor.objects.get(id=vendor_id) except Vendor.DoesNotExist: return JsonResponse({'error': 'Estabelecimento não encontrado.'}, status=404) user = request.api_user is_owner = _user_owns_object(user, 'vendor', vendor.id) if request.method == 'GET': # Also get aggregated data for this user at this vendor order_ids = _get_user_order_ids(user) agg = ( Order.objects.filter(id__in=order_ids, vendor_id=vendor.id) .aggregate( total_spent=Sum('total_paid'), order_count=Count('id'), ) ) return JsonResponse({ 'id': vendor.id, 'nome_fantasia': vendor.nomefantasia_text or '', 'razao_social': vendor.nomerazaosocial_text or '', 'cnpj': vendor.cpfcnpj_text or '', 'endereco': vendor.endereco_text or '', 'bairro': vendor.bairro_text or '', 'cep': vendor.cep_text or '', 'municipio': vendor.municipio_text or '', 'uf': vendor.uf_text or '', 'telefone': vendor.telefone_text or '', 'latitude': vendor.latitude, 'longitude': vendor.longitude, 'total_spent': round(agg['total_spent'] or 0, 2), 'order_count': agg['order_count'] or 0, 'can_edit': is_owner, }) if request.method == 'PATCH': if not is_owner: return JsonResponse({'error': 'Apenas o autor pode editar este estabelecimento.'}, status=403) try: body = json.loads(request.body) except (json.JSONDecodeError, ValueError): return JsonResponse({'error': 'JSON inválido.'}, status=400) nome = body.get('nome_fantasia', '').strip() if not nome: return JsonResponse({'error': 'nome_fantasia é obrigatório.'}, status=400) vendor.nomefantasia_text = nome vendor.save(update_fields=['nomefantasia_text']) return JsonResponse({ 'id': vendor.id, 'nome_fantasia': vendor.nomefantasia_text, }) return JsonResponse({'error': 'Method not allowed'}, status=405) @csrf_exempt @require_auth def vendor_orders(request, vendor_id): """ GET /api/estabelecimentos/:id/compras/ — Orders by the user at this vendor. """ if request.method != 'GET': return JsonResponse({'error': 'Method not allowed'}, status=405) try: vendor = Vendor.objects.get(id=vendor_id) except Vendor.DoesNotExist: return JsonResponse({'error': 'Estabelecimento não encontrado.'}, status=404) user = request.api_user order_ids = _get_user_order_ids(user) orders = ( Order.objects.filter(id__in=order_ids, vendor_id=vendor.id) .order_by('-sale_date') ) result = [] for o in orders: result.append({ 'id': o.id, 'sale_date': o.sale_date.isoformat() if o.sale_date else None, 'total_paid': round(o.total_paid, 2), 'total_discount': round(o.total_discount, 2), 'total_taxes': round(o.total_taxes, 2), 'payment_form': o.payment_form or '', 'total_quantity': o.total_quantity, }) return JsonResponse(result, safe=False) @csrf_exempt @require_auth def order_detail(request, order_id): """ GET /api/compras/:id/ — Order detail with list of products. PATCH /api/compras/:id/ — Update installment info for credit card payments. """ try: order = Order.objects.select_related('vendor').get(id=order_id) except Order.DoesNotExist: return JsonResponse({'error': 'Compra não encontrada.'}, status=404) # Ownership check user = request.api_user order_ids = _get_user_order_ids(user) if order.id not in order_ids: return JsonResponse({'error': 'Acesso negado.'}, status=403) def _is_credit_card_payment(text: str) -> bool: t = (text or '').lower() return ('cartão de crédito' in t) or ('cartao de credito' in t) or (t.strip() in ['cc', '3 - cartão de crédito', '3 - cartao de credito']) def _normalize_payment_form_str(raw: str) -> str: if not raw: return '' s = raw.strip() # Remove leading junk occurrences like "();" while s.startswith('();') or s.startswith('() ;') or s.startswith('() ;'): s = s[3:].lstrip() if s.startswith('()'): s = s[2:].lstrip(' ;') s = s.lstrip(';').strip() # Remove empty segments and segments that are just "()" parts = [p.strip() for p in s.split(';') if p.strip() and p.strip() != '()'] # Also strip "()" parts2 = [] for p in parts: if p.startswith('()'): p = p[2:].strip() if p: parts2.append(p) return ';'.join(parts2) def _parse_brl_to_float(text: str) -> float: if text is None: return 0.0 cleaned = re.sub(r'[^\d,.]', '', str(text)).strip() if not cleaned: return 0.0 if ',' in cleaned: cleaned = cleaned.replace('.', '').replace(',', '.') try: return float(cleaned) except ValueError: return 0.0 def _format_brl(value: float) -> str: # 100.67 -> "100,67" return f'{value:.2f}'.replace('.', ',') def _apply_installments_to_payment_string(payment_form: str, parcelas: int, total_paid: float) -> str: """ Transform "... Cartão de Crédito (302,03) ..." into "... Cartão de Crédito (3x100,67) ..." If the string doesn't have parentheses, append them to the credit card label. """ if not payment_form: return payment_form per_installment = round((total_paid or 0) / max(parcelas, 1), 2) token = f'{parcelas}x{_format_brl(per_installment)}' # Replace the first "(...)" that belongs to the credit card label. # Supports variants like "3 - Cartão de Crédito (302,03)". pattern = re.compile(r'((?:\d+\s*-\s*)?cart[aã]o\s+de\s+cr[eé]dito)\s*\([^)]*\)', re.IGNORECASE) if pattern.search(payment_form): return pattern.sub(rf'\1 ({token})', payment_form, count=1) # If no parentheses were found, add "(NxVALOR)" right after the label. pattern2 = re.compile(r'((?:\d+\s*-\s*)?cart[aã]o\s+de\s+cr[eé]dito)', re.IGNORECASE) if pattern2.search(payment_form): return pattern2.sub(rf'\1 ({token})', payment_form, count=1) return payment_form def _apply_installments_multi(payment_form: str, parcelas_list): """ Apply installments to each credit card segment. - If parcelas_list has N values, they will be applied in order of appearance to each "Cartão de Crédito (...)" segment. - The per-installment amount is derived from the segment's original value inside parentheses (preferred). If missing, falls back to order.total_paid. """ payment_form = _normalize_payment_form_str(payment_form) if not payment_form: return payment_form parts = [p.strip() for p in payment_form.split(';') if p.strip()] cc_pattern = re.compile(r'((?:\d+\s*-\s*)?cart[aã]o\s+de\s+cr[eé]dito)\s*(\([^)]*\))?', re.IGNORECASE) cc_idx = 0 out = [] for part in parts: m = cc_pattern.search(part) if not m: out.append(part) continue parcelas = parcelas_list[min(cc_idx, len(parcelas_list) - 1)] cc_idx += 1 # Try parse value inside parentheses as the base for per-installment calc paren = m.group(2) or '' base_val = _parse_brl_to_float(paren) if paren else 0.0 if base_val <= 0: base_val = float(total_paid or 0) per_installment = round(base_val / max(parcelas, 1), 2) token = f'{parcelas}x{_format_brl(per_installment)}' # Replace existing parentheses or append if paren: new_part = re.sub(r'\([^)]*\)', f'({token})', part, count=1) else: new_part = part + f' ({token})' out.append(new_part) return ';'.join(out) if request.method == 'PATCH': try: body = json.loads(request.body) except (json.JSONDecodeError, ValueError): return JsonResponse({'error': 'JSON inválido.'}, status=400) raw_parcelas = body.get('numero_parcelas') if raw_parcelas is None: return JsonResponse({'error': 'numero_parcelas é obrigatório.'}, status=400) parcelas_list = None if isinstance(raw_parcelas, list): parcelas_list = raw_parcelas elif isinstance(raw_parcelas, str): # Accept "3;2" or "3,2" for multi-card credit tokens = [t.strip() for t in re.split(r'[;,]', raw_parcelas) if t.strip()] parcelas_list = tokens if tokens else [raw_parcelas] else: parcelas_list = [raw_parcelas] parsed = [] for p in parcelas_list: try: n = int(str(p).strip()) except Exception: return JsonResponse({'error': 'numero_parcelas inválido.'}, status=400) if n < 1 or n > 24: return JsonResponse({'error': 'numero_parcelas deve estar entre 1 e 24.'}, status=400) parsed.append(n) parcelas_list = parsed order.payment_form = _normalize_payment_form_str(order.payment_form or '') if not _is_credit_card_payment(order.payment_form): return JsonResponse( {'error': 'Parcelamento só pode ser informado para Cartão de Crédito.'}, status=400, ) order.payment_form = _apply_installments_multi(order.payment_form, parcelas_list) order.contem_parcelamento = any(n > 1 for n in parcelas_list) # Keep a representative value (max) for compatibility; actual split is derived from payment_form when present. order.numero_parcelas = max(parcelas_list) if parcelas_list else 1 order.save(update_fields=['numero_parcelas', 'contem_parcelamento', 'payment_form']) return JsonResponse({ 'id': order.id, 'numero_parcelas': order.numero_parcelas, 'contem_parcelamento': order.contem_parcelamento, 'payment_form': order.payment_form or '', }) if request.method != 'GET': return JsonResponse({'error': 'Method not allowed'}, status=405) # Get products items = ( OrderProduct.objects.filter(order=order) .select_related('product') .order_by('id') ) products = [] for op in items: products.append({ 'id': op.product_id, 'description': op.product.description_text if op.product else '', 'ean': op.product.ean if op.product else '', 'quantity': round(op.product_quantity, 4), 'unit_price': round(op.product_un_price, 2), 'total_price': round(op.product_total_price, 2), 'taxes': round(op.product_total_taxes, 2), }) vendor = order.vendor payment_labels = dict(Order.PAYMENT_FORM) order.payment_form = _normalize_payment_form_str(order.payment_form or '') # Prefer the raw string when it already contains a human-readable label / values. payment_display = ( order.payment_form if (order.payment_form and ('(' in order.payment_form or '-' in order.payment_form or 'cart' in order.payment_form.lower())) else payment_labels.get(order.payment_form, order.payment_form or '') ) importer = _get_order_importer(order.id) return JsonResponse({ 'id': order.id, 'sale_date': order.sale_date.isoformat() if order.sale_date else None, 'total_price': round(order.total_price, 2), 'total_paid': round(order.total_paid, 2), 'total_discount': round(order.total_discount, 2), 'total_taxes': round(order.total_taxes, 2), 'payment_form': order.payment_form or '', 'payment_display': payment_display, 'contem_parcelamento': bool(getattr(order, 'contem_parcelamento', False)), 'numero_parcelas': int(getattr(order, 'numero_parcelas', 1) or 1), 'total_quantity': order.total_quantity, 'vendor': { 'id': vendor.id, 'nome_fantasia': vendor.nomefantasia_text or '', 'razao_social': vendor.nomerazaosocial_text or '', 'cnpj': vendor.cpfcnpj_text or '', }, 'imported_by_name': importer['name'] if importer else '', 'imported_by': importer, 'products': products, }) # --------------------------------------------------------------------------- # Import views # --------------------------------------------------------------------------- @csrf_exempt @require_POST @require_auth def import_sefaz(request): """ POST /api/import/ — Enqueue an NFC-e import job. Expects JSON body: { "key": "53260235377959..." } Validates the key, checks for duplicates, then creates an ImportJob record with status='pending'. A background worker picks it up and performs the actual SEFAZ scraping + DB persistence. Returns 202 Accepted with the job status immediately. """ try: body = json.loads(request.body) except (json.JSONDecodeError, ValueError): return JsonResponse({'error': 'JSON inválido.'}, status=400) key = body.get('key', '').strip() if not re.fullmatch(r'[0-9]{44}', key): return JsonResponse( {'error': 'Chave inválida. Deve conter exatamente 44 dígitos numéricos.'}, status=400, ) if not key.startswith(settings.CHAVE_PREFIX): return JsonResponse( { 'error': ( f'O Aplicativo {settings.SITE_NAME} funciona apenas ' f'para compras no {settings.SITE_REGION}.' ), }, status=400, ) if Order.objects.filter(key_sefaz=key).exists(): return JsonResponse( {'error': 'Esta nota fiscal já foi importada anteriormente.'}, status=409, ) if ImportJob.objects.filter(key_sefaz=key, status__in=['pending', 'processing']).exists(): return JsonResponse( {'error': 'Esta nota fiscal já está na fila de importação.'}, status=409, ) job = ImportJob.objects.create( key_sefaz=key, user=request.api_user, ) return JsonResponse({ 'job_id': job.id, 'key': job.key_sefaz, 'status': job.status, }, status=202) @csrf_exempt @require_GET @require_auth def import_status(request): """ GET /api/import/status/?keys=key1,key2,... — Batch status check. Returns the current status for each requested SEFAZ key, including result data for completed jobs and error messages for failed ones. """ raw_keys = request.GET.get('keys', '') keys = [k.strip() for k in raw_keys.split(',') if k.strip()] if not keys: return JsonResponse({'jobs': []}) user = request.api_user family_ids = _get_family_user_ids(user) jobs = ( ImportJob.objects .filter(key_sefaz__in=keys, user_id__in=family_ids) .order_by('-created_at') ) seen_keys = set() data = [] for job in jobs: if job.key_sefaz in seen_keys: continue seen_keys.add(job.key_sefaz) entry = { 'job_id': job.id, 'key': job.key_sefaz, 'status': job.status, 'created_at': job.created_at.isoformat(), } if job.status == 'completed' and job.result: entry['result'] = job.result if job.status == 'failed': entry['error'] = job.error_message or 'Erro desconhecido.' entry['retry_count'] = job.retry_count data.append(entry) already_imported = set() for key in keys: if key not in seen_keys and Order.objects.filter(key_sefaz=key).exists(): already_imported.add(key) for key in already_imported: data.append({ 'key': key, 'status': 'completed', }) return JsonResponse({'jobs': data}) # --------------------------------------------------------------------------- # Contas Básicas (água / luz) # --------------------------------------------------------------------------- def _conta_to_dict(conta): return { 'id': conta.id, 'tipo': conta.tipo, 'codigo_barras': conta.codigo_barras, 'data_vencimento': conta.data_vencimento.isoformat() if conta.data_vencimento else None, 'valor_devido': conta.valor_devido, 'valor_pago': conta.valor_pago, 'data_pagamento': conta.data_pagamento.isoformat() if conta.data_pagamento else None, } @csrf_exempt @require_auth def conta_list_create(request): """ GET /api/contas/ — List bills owned by the authenticated user. POST /api/contas/ — Create a new bill (+ LogEntry for ownership). """ if request.method == 'GET': return _conta_list(request) elif request.method == 'POST': return _conta_create(request) else: return JsonResponse({'error': 'Método não permitido.'}, status=405) def _conta_list(request): user = request.api_user family_ids = _get_family_user_ids(user) ct = ContentType.objects.get_for_model(ContaBasica) owned_ids = LogEntry.objects.filter( user_id__in=family_ids, action_flag=ADDITION, content_type_id=ct.id, ).values_list('object_id', flat=True) contas = ContaBasica.objects.filter(id__in=[int(i) for i in owned_ids]) return JsonResponse([_conta_to_dict(c) for c in contas], safe=False) def _conta_create(request): try: body = json.loads(request.body) except (json.JSONDecodeError, ValueError): return JsonResponse({'error': 'JSON inválido.'}, status=400) tipo = body.get('tipo', '').strip() if tipo not in ('agua', 'luz', 'iptu', 'claro', 'gas'): return JsonResponse({'error': "tipo deve ser 'agua', 'luz', 'iptu', 'claro' ou 'gas'."}, status=400) valor_devido = body.get('valor_devido') if valor_devido is None: return JsonResponse({'error': 'valor_devido é obrigatório.'}, status=400) try: valor_devido = float(valor_devido) except (TypeError, ValueError): return JsonResponse({'error': 'valor_devido inválido.'}, status=400) valor_pago = body.get('valor_pago') if valor_pago is not None: try: valor_pago = float(valor_pago) except (TypeError, ValueError): return JsonResponse({'error': 'valor_pago inválido.'}, status=400) codigo_barras = body.get('codigo_barras', '') data_vencimento = None raw_vencimento = body.get('data_vencimento') if raw_vencimento: try: data_vencimento = datetime.date.fromisoformat(raw_vencimento) except (ValueError, TypeError): return JsonResponse({'error': 'data_vencimento inválida (esperado YYYY-MM-DD).'}, status=400) data_pagamento = None raw_pagamento = body.get('data_pagamento') if raw_pagamento: try: data_pagamento = datetime.date.fromisoformat(raw_pagamento) except (ValueError, TypeError): return JsonResponse({'error': 'data_pagamento inválida (esperado YYYY-MM-DD).'}, status=400) conta = ContaBasica.objects.create( tipo=tipo, codigo_barras=codigo_barras, valor_devido=valor_devido, valor_pago=valor_pago, data_vencimento=data_vencimento, data_pagamento=data_pagamento, ) ct = ContentType.objects.get_for_model(ContaBasica) LogEntry.objects.create( user_id=request.api_user.id, content_type_id=ct.id, object_id=str(conta.id), object_repr=str(conta), action_flag=ADDITION, ) return JsonResponse(_conta_to_dict(conta), status=200) @csrf_exempt @require_auth def conta_detail(request, conta_id): """ GET /api/contas/:id/ — Conta detail. PATCH /api/contas/:id/ — Update conta fields (owner only). DELETE /api/contas/:id/ — Delete conta (owner only). """ try: conta = ContaBasica.objects.get(id=conta_id) except ContaBasica.DoesNotExist: return JsonResponse({'error': 'Conta não encontrada.'}, status=404) user = request.api_user if not _user_owns_object(user, 'contabasica', conta.id): return JsonResponse( {'error': 'Você não tem permissão para acessar esta conta.'}, status=403, ) if request.method == 'GET': return JsonResponse(_conta_to_dict(conta)) elif request.method == 'PATCH': return _conta_update(request, conta) elif request.method == 'DELETE': return _conta_delete(request, conta) else: return JsonResponse({'error': 'Método não permitido.'}, status=405) def _conta_update(request, conta): """Update allowed fields on a conta. Requires ownership (already checked).""" try: body = json.loads(request.body) except (json.JSONDecodeError, ValueError): return JsonResponse({'error': 'JSON inválido.'}, status=400) # valor_pago if 'valor_pago' in body: valor_pago = body['valor_pago'] if valor_pago is not None: try: valor_pago = float(valor_pago) except (TypeError, ValueError): return JsonResponse({'error': 'valor_pago inválido.'}, status=400) conta.valor_pago = valor_pago # data_vencimento if 'data_vencimento' in body: raw = body['data_vencimento'] if raw: try: conta.data_vencimento = datetime.date.fromisoformat(raw) except (ValueError, TypeError): return JsonResponse( {'error': 'data_vencimento inválida (esperado YYYY-MM-DD).'}, status=400, ) else: conta.data_vencimento = None # data_pagamento if 'data_pagamento' in body: raw = body['data_pagamento'] if raw: try: conta.data_pagamento = datetime.date.fromisoformat(raw) except (ValueError, TypeError): return JsonResponse( {'error': 'data_pagamento inválida (esperado YYYY-MM-DD).'}, status=400, ) else: conta.data_pagamento = None conta.save() return JsonResponse(_conta_to_dict(conta)) def _conta_delete(request, conta): """Delete the conta. Requires ownership (already checked).""" conta.delete() return JsonResponse({'detail': 'Conta excluída com sucesso.'}, status=200) @csrf_exempt @require_auth def conta_history(request, conta_id): """ GET /api/contas//historico/ — Price history for a conta. Returns stats (min, max, avg, count) and a chronological list of all contas of the same tipo owned by the authenticated user. """ if request.method != 'GET': return JsonResponse({'error': 'Método não permitido.'}, status=405) try: conta = ContaBasica.objects.get(id=conta_id) except ContaBasica.DoesNotExist: return JsonResponse({'error': 'Conta não encontrada.'}, status=404) user = request.api_user if not _user_owns_object(user, 'contabasica', conta.id): return JsonResponse( {'error': 'Você não tem permissão para acessar esta conta.'}, status=403, ) # Get all conta IDs owned by this user or family members family_ids = _get_family_user_ids(user) ct = ContentType.objects.get_for_model(ContaBasica) owned_ids = LogEntry.objects.filter( user_id__in=family_ids, action_flag=ADDITION, content_type_id=ct.id, ).values_list('object_id', flat=True) # Filter for the same tipo, ordered by data_vencimento same_tipo = ( ContaBasica.objects .filter(id__in=[int(i) for i in owned_ids], tipo=conta.tipo) .order_by('data_vencimento') ) # Build history list history = [] for c in same_tipo: history.append({ 'id': c.id, 'date': c.data_vencimento.isoformat() if c.data_vencimento else None, 'valor_pago': round(c.valor_pago, 2) if c.valor_pago is not None else None, 'valor_devido': round(c.valor_devido, 2) if c.valor_devido is not None else None, }) # Compute stats from valor_pago of the same tipo agg = same_tipo.aggregate( min_price=Min('valor_pago'), max_price=Max('valor_pago'), avg_price=Avg('valor_pago'), total_count=Count('id'), ) stats = { 'min_price': round(agg['min_price'], 2) if agg['min_price'] is not None else None, 'max_price': round(agg['max_price'], 2) if agg['max_price'] is not None else None, 'avg_price': round(agg['avg_price'], 2) if agg['avg_price'] is not None else None, 'total_count': agg['total_count'], } return JsonResponse({ 'tipo': conta.tipo, 'stats': stats, 'history': history, }) # --------------------------------------------------------------------------- # Family Group views # --------------------------------------------------------------------------- @csrf_exempt @require_auth def family_detail(request): """ GET /api/family/ — Return the user's family group, members, and pending invites. POST /api/family/ — Create a family group (if the user doesn't have one). DELETE /api/family/ — Delete the family group (creator only). """ user = request.api_user if request.method == 'GET': fg = FamilyGroup.objects.filter(group__user=user).select_related('group').first() # Also return invites received by this user received = FamilyInvite.objects.filter( invited_email__iexact=user.email, status='pending', ).select_related('family', 'invited_by') received_list = [] for inv in received: received_list.append({ 'id': inv.id, 'invited_by_name': inv.invited_by.get_full_name() or inv.invited_by.username, 'invited_by_email': inv.invited_by.email, 'created_at': inv.created_at.isoformat(), }) if not fg: return JsonResponse({ 'group': None, 'received_invites': received_list, }) members = [] for m in fg.group.user_set.all(): members.append({ 'id': m.id, 'name': m.get_full_name() or m.username, 'email': m.email, 'is_creator': m.id == fg.created_by_id, }) pending_invites = [] for inv in fg.invites.filter(status='pending'): pending_invites.append({ 'id': inv.id, 'invited_email': inv.invited_email, 'created_at': inv.created_at.isoformat(), }) return JsonResponse({ 'group': { 'id': fg.id, 'created_at': fg.created_at.isoformat(), 'is_creator': user.id == fg.created_by_id, 'members': members, 'pending_invites': pending_invites, }, 'received_invites': received_list, }) if request.method == 'POST': if FamilyGroup.objects.filter(group__user=user).exists(): return JsonResponse({'error': 'Você já pertence a um grupo familiar.'}, status=400) from django.contrib.auth.models import Group as AuthGroup group_name = f'family_{user.id}_{int(timezone.now().timestamp())}' auth_group = AuthGroup.objects.create(name=group_name) auth_group.user_set.add(user) fg = FamilyGroup.objects.create(group=auth_group, created_by=user) return JsonResponse({ 'id': fg.id, 'created_at': fg.created_at.isoformat(), }, status=201) if request.method == 'DELETE': fg = FamilyGroup.objects.filter(created_by=user).select_related('group').first() if not fg: return JsonResponse({'error': 'Você não é o criador de nenhum grupo familiar.'}, status=403) auth_group = fg.group fg.delete() auth_group.delete() return JsonResponse({'detail': 'Grupo familiar excluído com sucesso.'}) return JsonResponse({'error': 'Método não permitido.'}, status=405) @csrf_exempt @require_auth def family_invite(request): """ POST /api/family/invite/ — Invite a user by email. Expects JSON body: { "email": "someone@example.com" } Creates the group automatically if the user doesn't have one yet. """ if request.method != 'POST': return JsonResponse({'error': 'Método não permitido.'}, status=405) user = request.api_user try: body = json.loads(request.body) except (json.JSONDecodeError, ValueError): return JsonResponse({'error': 'JSON inválido.'}, status=400) email = body.get('email', '').strip().lower() if not email: return JsonResponse({'error': 'E-mail é obrigatório.'}, status=400) if email == user.email.lower(): return JsonResponse({'error': 'Você não pode convidar a si mesmo.'}, status=400) # Get or create the family group fg = FamilyGroup.objects.filter(group__user=user).select_related('group').first() if not fg: from django.contrib.auth.models import Group as AuthGroup group_name = f'family_{user.id}_{int(timezone.now().timestamp())}' auth_group = AuthGroup.objects.create(name=group_name) auth_group.user_set.add(user) fg = FamilyGroup.objects.create(group=auth_group, created_by=user) # Only the creator can invite if fg.created_by_id != user.id: return JsonResponse({'error': 'Apenas o criador do grupo pode convidar membros.'}, status=403) # Check if email is already a member if fg.group.user_set.filter(email__iexact=email).exists(): return JsonResponse({'error': 'Este usuário já é membro do grupo.'}, status=400) # Check for existing pending invite if FamilyInvite.objects.filter(family=fg, invited_email__iexact=email, status='pending').exists(): return JsonResponse({'error': 'Já existe um convite pendente para este e-mail.'}, status=400) invite = FamilyInvite.objects.create( family=fg, invited_email=email, invited_by=user, ) # Create in-app notification + push for the invited user (if they exist) inviter_name = user.get_full_name() or user.username try: invited_user = User.objects.get(email__iexact=email) notif = Notification.objects.create( user=invited_user, title='Convite Familiar', body=f'{inviter_name} te convidou para o grupo familiar. Toque para aceitar ou recusar.', ) # Send push notification try: profile = UserProfile.objects.get(user=invited_user) if profile.expo_push_token: send_push_notification( profile.expo_push_token, 'Convite Familiar', f'{inviter_name} te convidou para o grupo familiar.', data={ 'notification_id': notif.id, 'type': 'family_invite', }, ) except UserProfile.DoesNotExist: pass except User.DoesNotExist: pass return JsonResponse({ 'id': invite.id, 'invited_email': invite.invited_email, 'status': invite.status, 'created_at': invite.created_at.isoformat(), }, status=201) @csrf_exempt @require_auth def family_invites(request): """ GET /api/family/invites/ — List pending invites received by the authenticated user. """ if request.method != 'GET': return JsonResponse({'error': 'Método não permitido.'}, status=405) user = request.api_user invites = FamilyInvite.objects.filter( invited_email__iexact=user.email, status='pending', ).select_related('invited_by', 'family') result = [] for inv in invites: result.append({ 'id': inv.id, 'invited_by_name': inv.invited_by.get_full_name() or inv.invited_by.username, 'invited_by_email': inv.invited_by.email, 'created_at': inv.created_at.isoformat(), }) return JsonResponse(result, safe=False) @csrf_exempt @require_auth def family_invite_accept(request, invite_id): """ POST /api/family/invites//accept/ — Accept a pending invite. """ if request.method != 'POST': return JsonResponse({'error': 'Método não permitido.'}, status=405) user = request.api_user try: invite = FamilyInvite.objects.select_related('family__group').get( id=invite_id, invited_email__iexact=user.email, status='pending', ) except FamilyInvite.DoesNotExist: return JsonResponse({'error': 'Convite não encontrado ou já processado.'}, status=404) # Check if user is already in another family group if FamilyGroup.objects.filter(group__user=user).exists(): return JsonResponse( {'error': 'Você já pertence a um grupo familiar. Saia primeiro para aceitar este convite.'}, status=400, ) invite.status = 'accepted' invite.save(update_fields=['status']) invite.family.group.user_set.add(user) return JsonResponse({'detail': 'Convite aceito com sucesso.'}) @csrf_exempt @require_auth def family_invite_reject(request, invite_id): """ POST /api/family/invites//reject/ — Reject a pending invite. """ if request.method != 'POST': return JsonResponse({'error': 'Método não permitido.'}, status=405) user = request.api_user try: invite = FamilyInvite.objects.get( id=invite_id, invited_email__iexact=user.email, status='pending', ) except FamilyInvite.DoesNotExist: return JsonResponse({'error': 'Convite não encontrado ou já processado.'}, status=404) invite.status = 'rejected' invite.save(update_fields=['status']) return JsonResponse({'detail': 'Convite recusado.'}) @csrf_exempt @require_auth def family_remove_member(request, user_id): """ DELETE /api/family/members// — Remove a member (creator only) or leave the group (if user_id is the authenticated user). """ if request.method != 'DELETE': return JsonResponse({'error': 'Método não permitido.'}, status=405) user = request.api_user fg = FamilyGroup.objects.filter(group__user=user).select_related('group').first() if not fg: return JsonResponse({'error': 'Você não pertence a nenhum grupo familiar.'}, status=404) try: target_user = User.objects.get(id=user_id) except User.DoesNotExist: return JsonResponse({'error': 'Usuário não encontrado.'}, status=404) if not fg.group.user_set.filter(id=target_user.id).exists(): return JsonResponse({'error': 'Este usuário não é membro do grupo.'}, status=400) # Creator can remove anyone except themselves via this endpoint if user.id == fg.created_by_id: if target_user.id == user.id: return JsonResponse( {'error': 'O criador não pode se remover. Exclua o grupo em vez disso.'}, status=400, ) fg.group.user_set.remove(target_user) return JsonResponse({'detail': 'Membro removido com sucesso.'}) # Non-creator can only remove themselves (leave) if target_user.id != user.id: return JsonResponse({'error': 'Apenas o criador pode remover outros membros.'}, status=403) fg.group.user_set.remove(user) return JsonResponse({'detail': 'Você saiu do grupo familiar.'}) # --------------------------------------------------------------------------- # Dashboard views # --------------------------------------------------------------------------- @csrf_exempt @require_GET @require_auth def dashboard_data(request): """ GET /api/dashboard/ — Aggregated dashboard data for a calendar month. Optional query param: month=YYYY-MM (defaults to the current month). """ user = request.api_user family_ids = _get_family_user_ids(user) now = timezone.now() current_month_start = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0) month_param = (request.GET.get('month') or '').strip() if month_param: try: parsed = datetime.datetime.strptime(month_param, '%Y-%m') month_start = current_month_start.replace( year=parsed.year, month=parsed.month, day=1, ) except ValueError: return JsonResponse( {'error': 'Parâmetro month inválido. Use YYYY-MM.'}, status=400, ) else: month_start = current_month_start last_day = calendar.monthrange(month_start.year, month_start.month)[1] month_end_date = month_start.date().replace(day=last_day) contas_end_date = ( min(month_end_date, now.date()) if month_start == current_month_start else month_end_date ) # --- User settings --- try: profile = UserProfile.objects.get(user=user) monthly_income = profile.monthly_income monthly_goal = profile.monthly_goal except UserProfile.DoesNotExist: monthly_income = None monthly_goal = None # --- Order IDs for the family --- order_ids = _get_user_order_ids(user) def _month_key(dt): return dt.strftime('%Y-%m') # --- Orders in scope for installment distribution --- # We need orders from previous months too, because their installments may fall into the selected month. # Use last 24 months as a safe bound (parcelas limited to 24 via PATCH, but typical is small). orders_for_distribution = ( Order.objects .filter(id__in=order_ids, sale_date__gte=add_months(month_start, -24)) .select_related('vendor') ) # --- Meta vs Gastos (selected month) --- total_spent_orders = 0.0 for o in orders_for_distribution: total_spent_orders += order_installment_amount_for_month(o, month_start) ct_conta = ContentType.objects.get_for_model(ContaBasica) owned_conta_ids = LogEntry.objects.filter( user_id__in=family_ids, action_flag=ADDITION, content_type_id=ct_conta.id, ).values_list('object_id', flat=True) contas_this_month = ContaBasica.objects.filter( id__in=[int(i) for i in owned_conta_ids], data_pagamento__gte=month_start.date(), data_pagamento__lte=contas_end_date, ) total_spent_contas = contas_this_month.aggregate( total=Sum('valor_pago'), )['total'] or 0 total_spent = round(total_spent_orders + total_spent_contas, 2) # --- Gastos por Loja (top 5 this month) --- vendor_totals = {} for o in orders_for_distribution: amt = order_installment_amount_for_month(o, month_start) if amt <= 0: continue vendor_totals[o.vendor_id] = vendor_totals.get(o.vendor_id, 0.0) + amt top_vendor_rows = sorted( [{'vendor_id': vid, 'total': tot} for vid, tot in vendor_totals.items()], key=lambda x: x['total'], reverse=True, )[:5] vendor_ids = [v['vendor_id'] for v in top_vendor_rows] vendor_map = {v.id: v for v in Vendor.objects.filter(id__in=vendor_ids)} gastos_por_loja = [] for row in top_vendor_rows: v = vendor_map.get(row['vendor_id']) if v: gastos_por_loja.append({ 'vendor_id': v.id, 'name': v.nomefantasia_text or v.nomerazaosocial_text or v.cpfcnpj_text, 'total': round(row['total'] or 0, 2), }) # --- Gastos por Categoria (this month) --- # Only order products whose order has a non-zero installment amount in this month. orders_month_amount = {} for o in orders_for_distribution: orders_month_amount[o.id] = order_installment_amount_for_month(o, month_start) relevant_order_ids = [oid for oid, amt in orders_month_amount.items() if amt > 0] order_products_month = ( OrderProduct.objects .filter(order_id__in=relevant_order_ids) .select_related('product', 'order') ) category_totals = {} for op in order_products_month: order_amt = orders_month_amount.get(op.order_id, 0.0) if order_amt <= 0: continue o = op.order total_paid = float(getattr(o, 'total_paid', 0) or 0) factor = (order_amt / total_paid) if total_paid > 0 else 0.0 cat_id, cat_name = categorize_ncm(op.product.ncm) if cat_id not in category_totals: category_totals[cat_id] = {'id': cat_id, 'name': cat_name, 'total': 0} category_totals[cat_id]['total'] += (op.product_total_price * factor) gastos_por_categoria = sorted( [{'id': v['id'], 'name': v['name'], 'total': round(v['total'], 2)} for v in category_totals.values()], key=lambda x: x['total'], reverse=True, ) # --- Resumo Contas (this month) --- resumo_contas = [] contas_by_type = ( contas_this_month .values('tipo') .annotate(total=Sum('valor_pago'), count=Count('id')) ) for row in contas_by_type: resumo_contas.append({ 'tipo': row['tipo'], 'total': round(row['total'] or 0, 2), 'count': row['count'], }) # --- Ultimas Compras (5 most recent) --- recent_orders = ( Order.objects.filter(id__in=order_ids) .select_related('vendor') .order_by('-sale_date')[:5] ) ultimas_compras = [] for o in recent_orders: ultimas_compras.append({ 'id': o.id, 'vendor_name': o.vendor.nomefantasia_text or o.vendor.nomerazaosocial_text or '', 'total_paid': round(o.total_paid, 2), 'sale_date': o.sale_date.isoformat() if o.sale_date else None, 'total_quantity': o.total_quantity, }) # --- Alertas de Preco (products with >20% deviation from average) --- product_ids = _get_user_product_ids(user) alertas_preco = [] stats_qs = ( OrderProduct.objects .filter(product_id__in=product_ids, order_id__in=order_ids) .values('product_id') .annotate( avg_price=Avg('product_un_price'), last_price=Max('product_un_price'), count=Count('id'), ) .filter(count__gte=2) ) product_map = {p.id: p for p in Product.objects.filter(id__in=product_ids)} for row in stats_qs: avg_p = row['avg_price'] if not avg_p or avg_p == 0: continue # Get the most recent price for this product latest_op = ( OrderProduct.objects .filter(product_id=row['product_id'], order_id__in=order_ids) .select_related('order') .order_by('-order__sale_date') .first() ) if not latest_op: continue latest_price = latest_op.product_un_price pct_change = ((latest_price - avg_p) / avg_p) * 100 if abs(pct_change) >= 20: prod = product_map.get(row['product_id']) if prod: alertas_preco.append({ 'product_id': prod.id, 'product_name': prod.description_text, 'avg_price': round(avg_p, 2), 'latest_price': round(latest_price, 2), 'pct_change': round(pct_change, 1), }) alertas_preco.sort(key=lambda x: abs(x['pct_change']), reverse=True) alertas_preco = alertas_preco[:10] # --- Tendencia Mensal (last 6 months, always anchored to current month) --- six_months_ago = (now.replace(day=1) - datetime.timedelta(days=180)).replace(day=1) # Build trend map by distributing installment amounts per month. trend_map = {} start_trend_month = six_months_ago.replace(day=1, hour=0, minute=0, second=0, microsecond=0) end_trend_month = current_month_start # Consider orders as far back as 24 months for installment spillover. orders_for_trend = ( Order.objects .filter(id__in=order_ids, sale_date__gte=add_months(start_trend_month, -24)) ) # Precompute the 6 months range trend_months = [] for i in range(6): trend_months.append(add_months(end_trend_month, -(5 - i))) for o in orders_for_trend: for mdt in trend_months: amt = order_installment_amount_for_month(o, mdt) if amt: key = _month_key(mdt) trend_map[key] = trend_map.get(key, 0.0) + amt monthly_contas = ( ContaBasica.objects.filter( id__in=[int(i) for i in owned_conta_ids], data_pagamento__gte=six_months_ago.date(), ) .annotate(month=TruncMonth('data_pagamento')) .values('month') .annotate(total=Sum('valor_pago')) .order_by('month') ) for row in monthly_contas: key = row['month'].strftime('%Y-%m') trend_map[key] = trend_map.get(key, 0) + (row['total'] or 0) tendencia_mensal = [] for i in range(6): dt = add_months(current_month_start, -(5 - i)) key = dt.strftime('%Y-%m') tendencia_mensal.append({ 'month': key, 'label': dt.strftime('%b'), 'total': round(trend_map.get(key, 0), 2), }) return JsonResponse({ 'meta_vs_gastos': { 'monthly_income': monthly_income, 'monthly_goal': monthly_goal, 'total_spent': total_spent, 'total_orders': round(total_spent_orders, 2), 'total_contas': round(total_spent_contas, 2), }, 'gastos_por_loja': gastos_por_loja, 'gastos_por_categoria': gastos_por_categoria, 'resumo_contas': resumo_contas, 'ultimas_compras': ultimas_compras, 'alertas_preco': alertas_preco, 'tendencia_mensal': tendencia_mensal, }) @csrf_exempt @require_http_methods(['GET', 'PATCH']) @require_auth def dashboard_settings(request): """ GET /api/dashboard/settings/ — Return user's dashboard settings. PATCH /api/dashboard/settings/ — Update monthly_income and/or monthly_goal. """ user = request.api_user profile, _ = UserProfile.objects.get_or_create(user=user) if request.method == 'GET': return JsonResponse({ 'monthly_income': profile.monthly_income, 'monthly_goal': profile.monthly_goal, }) if request.method == 'PATCH': try: body = json.loads(request.body) except (json.JSONDecodeError, ValueError): return JsonResponse({'error': 'JSON inválido.'}, status=400) if 'monthly_income' in body: val = body['monthly_income'] profile.monthly_income = float(val) if val is not None else None if 'monthly_goal' in body: val = body['monthly_goal'] profile.monthly_goal = float(val) if val is not None else None profile.save(update_fields=['monthly_income', 'monthly_goal']) return JsonResponse({ 'monthly_income': profile.monthly_income, 'monthly_goal': profile.monthly_goal, }) return JsonResponse({'error': 'Método não permitido.'}, status=405)