Browse Source

Add geolocation package with API endpoints and models

- Introduced a new geolocation package containing APIs for IP-based geolocation, reverse geocoding, and region information.
- Added URL routes for geolocation endpoints in the account app.
- Implemented models, serializers, and views for handling geolocation data.
- Included GeoLite2 databases for IP to city and country lookups.
- Enhanced location history updates with optimized batch processing for improved performance.
- Updated settings to include the new geolocation package in installed apps.
master
Mohsen Taba 4 months ago
parent
commit
cab79d676d
  1. 6
      apps/account/urls.py
  2. 3
      apps/account/views/location_history.py
  3. 194
      apps/geolocation_package/README.md
  4. 19
      apps/geolocation_package/__init__.py
  5. 6
      apps/geolocation_package/apps.py
  6. BIN
      apps/geolocation_package/data/GeoLite2-City.mmdb
  7. BIN
      apps/geolocation_package/data/GeoLite2-Country.mmdb
  8. 53
      apps/geolocation_package/migrations/0001_initial.py
  9. 27
      apps/geolocation_package/migrations/0002_rename_geonamescit_latitu_latitude_7e2a6d_idx_geonames_ci_latitud_443791_idx_and_more.py
  10. 1
      apps/geolocation_package/migrations/__init__.py
  11. 1
      apps/geolocation_package/models/__init__.py
  12. 24
      apps/geolocation_package/models/geoNames.py
  13. 1
      apps/geolocation_package/serializers/__init__.py
  14. 42
      apps/geolocation_package/serializers/geolocation.py
  15. 10
      apps/geolocation_package/utils/__init__.py
  16. 451
      apps/geolocation_package/utils/city_detection_ip.py
  17. 257
      apps/geolocation_package/utils/geo.py
  18. 2
      apps/geolocation_package/views/__init__.py
  19. 278
      apps/geolocation_package/views/geolocation.py
  20. 186
      apps/geolocation_package/views/region_info.py
  21. 1
      config/settings/base.py

6
apps/account/urls.py

@ -4,9 +4,13 @@ from rest_framework.routers import DefaultRouter
from apps.account import views from apps.account import views
from apps.geolocation_package.views.geolocation import IPGeolocationAPIView, ReverseGeolocationAPIView
from apps.geolocation_package.views.region_info import RegionInfoView
urlpatterns = [ urlpatterns = [
path('geolocation/', IPGeolocationAPIView.as_view(), name='ip-geo'),
path('geolocation/reverse/', ReverseGeolocationAPIView.as_view(), name='reverse-geo'),
path('region-info/', RegionInfoView.as_view(), name='region-info'),
# URL for user registration, accepts POST requests for creating new user instances. # URL for user registration, accepts POST requests for creating new user instances.
path('register/', views.UserRegisterView.as_view(), name='user-register'), path('register/', views.UserRegisterView.as_view(), name='user-register'),

3
apps/account/views/location_history.py

@ -10,7 +10,8 @@ from rest_framework.response import Response
from rest_framework.views import APIView from rest_framework.views import APIView
from rest_framework import status from rest_framework import status
from apps.account.models import LocationHistory from apps.account.models import LocationHistory
from apps.account.serializers import LocationHistorySerializer, ReverseGeolocationSerializer, ReverseGeolocationResponseSerializer
from apps.account.serializers import LocationHistorySerializer
from apps.geolocation_package.serializers import ReverseGeolocationSerializer, ReverseGeolocationResponseSerializer
import geoip2.database import geoip2.database
import geoip2.errors import geoip2.errors
from city_detection_ip import get_location_by_coordinates, get_location_by_ip, SPECIAL_COORDINATES from city_detection_ip import get_location_by_coordinates, get_location_by_ip, SPECIAL_COORDINATES

194
apps/geolocation_package/README.md

@ -0,0 +1,194 @@
# Geolocation Package
این پکیج شامل تمام فایل‌های مرتبط با 3 API زیر است که از پروژه اصلی استخراج شده‌اند:
## API Endpoints
| API | URL | View | Description |
|-----|-----|------|-------------|
| IP Geolocation | `geolocation/` | `IPGeolocationAPIView` | دریافت اطلاعات موقعیت مکانی بر اساس IP کاربر |
| Reverse Geolocation | `geolocation/coordinates/` | `ReverseGeolocationAPIView` | دریافت اطلاعات شهر/کشور بر اساس مختصات جغرافیایی |
| Region Info | `auth/user/region/` | `RegionInfoView` | دریافت اطلاعات منطقه کاربر (شامل browser detection) |
---
## 📁 ساختار پوشه‌ها
```
geolocation_package/
├── __init__.py
├── README.md
├── views/
│ ├── __init__.py
│ ├── geolocation.py # IPGeolocationAPIView, ReverseGeolocationAPIView
│ └── region_info.py # RegionInfoView + detect_browser_from_user_agent
├── serializers/
│ ├── __init__.py
│ └── geolocation.py # IPGeolocationSerializer, ReverseGeolocationSerializer, ReverseGeolocationResponseSerializer
├── models/
│ ├── __init__.py
│ └── geoNames.py # GeoNamesCity model
├── utils/
│ ├── __init__.py
│ ├── city_detection_ip.py # get_location_by_coordinates, get_location_by_ip
│ └── geo.py # توابع کمکی برای geocoding از API‌های مختلف
└── data/
├── GeoLite2-City.mmdb # MaxMind GeoIP2 City database (61.5 MB)
├── GeoLite2-Country.mmdb # MaxMind GeoIP2 Country database (9.2 MB)
└── geonames_city.sqlite # SQLite export of GeoNamesCity (628.6 MB, 5,115,708 records)
```
---
## 📋 فایل‌های استخراج شده
### Views (ویوها)
| فایل | مبدا | توضیحات |
|------|------|---------|
| `views/geolocation.py` | `apps/account/views/geolocation.py` | شامل `IPGeolocationAPIView` و `ReverseGeolocationAPIView` |
| `views/region_info.py` | `apps/account/views/location_history.py` (lines 138-248) | شامل `RegionInfoView` و `detect_browser_from_user_agent` |
### Serializers (سریالایزرها)
| فایل | مبدا | توضیحات |
|------|------|---------|
| `serializers/geolocation.py` | `apps/account/serializer/geolocation.py` | شامل `IPGeolocationSerializer`, `ReverseGeolocationSerializer`, `ReverseGeolocationResponseSerializer` |
### Models (مدل‌ها)
| فایل | مبدا | توضیحات |
|------|------|---------|
| `models/geoNames.py` | `apps/account/models/geoNames.py` | مدل `GeoNamesCity` با 5,115,708 رکورد شهر از سراسر جهان |
### Utils (ابزارها)
| فایل | مبدا | توضیحات |
|------|------|---------|
| `utils/city_detection_ip.py` | `city_detection_ip.py` | شامل `get_location_by_coordinates`, `get_location_by_ip`, `SPECIAL_COORDINATES` |
| `utils/geo.py` | `utils/geo.py` | توابع fallback برای geocoding از API‌های مختلف (Nominatim, BigDataCloud, etc.) |
### Data Files (فایل‌های داده)
| فایل | مبدا | سایز | توضیحات |
|------|------|------|---------|
| `data/GeoLite2-City.mmdb` | `utils/country_city_db/GeoLite2-City.mmdb` | 61.5 MB | دیتابیس MaxMind برای IP to City |
| `data/GeoLite2-Country.mmdb` | `utils/country_city_db/GeoLite2-Country.mmdb` | 9.2 MB | دیتابیس MaxMind برای IP to Country |
| `data/geonames_city.sqlite` | از PostgreSQL (GeoNamesCity model) | 628.6 MB | SQLite export با 5,115,708 رکورد |
---
## 🔧 وابستگی‌ها (Dependencies)
```txt
geoip2>=4.0.0
djangorestframework>=3.14.0
drf-yasg>=1.21.0
```
---
## 💾 SQLite Database Schema
جدول `geonames_city`:
```sql
CREATE TABLE geonames_city (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
country_code TEXT NOT NULL,
latitude REAL NOT NULL,
longitude REAL NOT NULL,
feature_class TEXT NOT NULL,
population INTEGER
);
-- Indexes
CREATE INDEX idx_lat_lon ON geonames_city (latitude, longitude);
CREATE INDEX idx_country_code ON geonames_city (country_code);
CREATE INDEX idx_feature_class ON geonames_city (feature_class);
CREATE INDEX idx_feature_lat_lon ON geonames_city (feature_class, latitude, longitude);
```
---
## 🚀 نحوه استفاده در پروژه جدید
### 1. کپی فولدر به پروژه
```bash
cp -r geolocation_package /path/to/new_project/apps/
```
### 2. تنظیم urls.py
```python
from apps.geolocation_package.views import (
IPGeolocationAPIView,
ReverseGeolocationAPIView,
RegionInfoView
)
urlpatterns = [
path('geolocation/', IPGeolocationAPIView.as_view(), name='ip-geolocation'),
path('geolocation/coordinates/', ReverseGeolocationAPIView.as_view(), name='geolocation-by-coordinates'),
path('auth/user/region/', RegionInfoView.as_view(), name='region-info'),
]
```
### 3. تنظیم مسیر دیتابیس
مسیر فایل‌های MMDB را در کد به محل جدید تغییر دهید:
```python
# در views/geolocation.py و views/region_info.py
CITY_DB_PATH = Path("apps/geolocation_package/data/GeoLite2-City.mmdb")
```
### 4. استفاده از SQLite به جای PostgreSQL (اختیاری)
اگر نمی‌خواهید GeoNamesCity را در PostgreSQL نگه دارید، می‌توانید مستقیماً از SQLite استفاده کنید:
```python
import sqlite3
from pathlib import Path
DB_PATH = Path("apps/geolocation_package/data/geonames_city.sqlite")
def get_city_by_coordinates(lat, lon):
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
cursor.execute("""
SELECT name, country_code FROM geonames_city
WHERE feature_class = 'P'
AND latitude BETWEEN ? AND ?
AND longitude BETWEEN ? AND ?
LIMIT 1
""", (lat - 0.5, lat + 0.5, lon - 0.5, lon + 0.5))
result = cursor.fetchone()
conn.close()
return result
```
---
## 📊 آمار دیتابیس
- **تعداد رکوردها در GeoNamesCity**: 5,115,708
- **حجم SQLite**: 628.59 MB
- **حجم GeoLite2-City.mmdb**: 61.50 MB
- **حجم GeoLite2-Country.mmdb**: 9.22 MB
---
## 📝 نکات مهم
1. **دیتابیس اصلی حذف نشده است** - فقط داده‌ها کپی شده‌اند
2. **GeoLite2 License**: فایل‌های MMDB از MaxMind هستند و نیاز به رعایت لایسنس دارند
3. **آپدیت دیتابیس**: برای آپدیت GeoNames از اسکریپت `export_geonames_to_sqlite.py` استفاده کنید
---
*Generated on: 2026-01-27*

19
apps/geolocation_package/__init__.py

@ -0,0 +1,19 @@
"""
Geolocation Package
A standalone package containing geolocation APIs for:
- IP-based geolocation (IPGeolocationAPIView)
- Reverse geocoding from coordinates (ReverseGeolocationAPIView)
- Region info with browser detection (RegionInfoView)
Models:
- GeoNamesCity: 5,115,708 cities worldwide
Data Files:
- GeoLite2-City.mmdb: MaxMind IP to City database
- GeoLite2-Country.mmdb: MaxMind IP to Country database
- geonames_city.sqlite: SQLite export of GeoNamesCity model
"""
__version__ = '1.0.0'
default_app_config = 'apps.geolocation_package.apps.GeolocationPackageConfig'

6
apps/geolocation_package/apps.py

@ -0,0 +1,6 @@
from django.apps import AppConfig
class GeolocationPackageConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'apps.geolocation_package' # Must match the path in INSTALLED_APPS
verbose_name = 'Geolocation Package'

BIN
apps/geolocation_package/data/GeoLite2-City.mmdb

BIN
apps/geolocation_package/data/GeoLite2-Country.mmdb

53
apps/geolocation_package/migrations/0001_initial.py

@ -0,0 +1,53 @@
# Generated manually for geolocation_package app
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = [
]
operations = [
migrations.CreateModel(
name='GeoNamesCity',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('name', models.CharField(max_length=200)),
('country_code', models.CharField(max_length=2)),
('latitude', models.FloatField()),
('longitude', models.FloatField()),
('feature_class', models.CharField(max_length=1)),
('population', models.BigIntegerField(null=True)),
],
options={
'db_table': 'geonames_city',
},
),
migrations.AddIndex(
model_name='geonamescity',
index=models.Index(fields=['latitude', 'longitude'], name='geonamescit_latitu_latitude_7e2a6d_idx'),
),
migrations.AddIndex(
model_name='geonamescity',
index=models.Index(fields=['country_code'], name='geonamescit_countr_country_c_5c8b6a_idx'),
),
migrations.AddIndex(
model_name='geonamescity',
index=models.Index(fields=['feature_class'], name='geonamescit_featur_feature_c_9b2a4d_idx'),
),
migrations.AddIndex(
model_name='geonamescity',
index=models.Index(fields=['feature_class', 'latitude', 'longitude'], name='idx_geonames_feature_lat_lon'),
),
migrations.AddIndex(
model_name='geonamescity',
index=models.Index(fields=['latitude'], name='idx_geonames_lat_populated', condition=models.Q(feature_class='P')),
),
migrations.AddIndex(
model_name='geonamescity',
index=models.Index(fields=['longitude'], name='idx_geonames_lon_populated', condition=models.Q(feature_class='P')),
),
]

27
apps/geolocation_package/migrations/0002_rename_geonamescit_latitu_latitude_7e2a6d_idx_geonames_ci_latitud_443791_idx_and_more.py

@ -0,0 +1,27 @@
# Generated by Django 4.2.27 on 2026-01-27 14:01
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
("geolocation_package", "0001_initial"),
]
operations = [
migrations.RenameIndex(
model_name="geonamescity",
new_name="geonames_ci_latitud_443791_idx",
old_name="geonamescit_latitu_latitude_7e2a6d_idx",
),
migrations.RenameIndex(
model_name="geonamescity",
new_name="geonames_ci_country_9c873a_idx",
old_name="geonamescit_countr_country_c_5c8b6a_idx",
),
migrations.RenameIndex(
model_name="geonamescity",
new_name="geonames_ci_feature_85b0fe_idx",
old_name="geonamescit_featur_feature_c_9b2a4d_idx",
),
]

1
apps/geolocation_package/migrations/__init__.py

@ -0,0 +1 @@
# This file makes Python treat the directory as a package

1
apps/geolocation_package/models/__init__.py

@ -0,0 +1 @@
from .geoNames import GeoNamesCity

24
apps/geolocation_package/models/geoNames.py

@ -0,0 +1,24 @@
from django.db import models
class GeoNamesCity(models.Model):
name = models.CharField(max_length=200)
country_code = models.CharField(max_length=2)
latitude = models.FloatField()
longitude = models.FloatField()
feature_class = models.CharField(max_length=1)
population = models.BigIntegerField(null=True)
class Meta:
indexes = [
models.Index(fields=['latitude', 'longitude']),
models.Index(fields=['country_code']),
models.Index(fields=['feature_class']),
# ایندکس بهینه برای کوری‌های جستجوی مکان
models.Index(fields=['feature_class', 'latitude', 'longitude'], name='idx_geonames_feature_lat_lon'),
# ایندکس‌های جداگانه برای محدوده جغرافیایی
models.Index(fields=['latitude'], condition=models.Q(feature_class='P'), name='idx_geonames_lat_populated'),
models.Index(fields=['longitude'], condition=models.Q(feature_class='P'), name='idx_geonames_lon_populated'),
]
db_table = 'geonames_city'
def __str__(self):
return f"{self.name}, {self.country_code}"

1
apps/geolocation_package/serializers/__init__.py

@ -0,0 +1 @@
from .geolocation import IPGeolocationSerializer, ReverseGeolocationSerializer, ReverseGeolocationResponseSerializer

42
apps/geolocation_package/serializers/geolocation.py

@ -0,0 +1,42 @@
from rest_framework import serializers
class IPGeolocationSerializer(serializers.Serializer):
"""Serializer for IP geolocation response"""
ip = serializers.IPAddressField(read_only=True, required=False)
country = serializers.CharField(max_length=100, allow_null=True, allow_blank=True, read_only=True, required=False)
country_code = serializers.CharField(max_length=10, allow_null=True, allow_blank=True, read_only=True, required=False)
city = serializers.CharField(max_length=100, allow_null=True, allow_blank=True, read_only=True, required=False)
latitude = serializers.FloatField(allow_null=True, read_only=True, required=False)
longitude = serializers.FloatField(allow_null=True, read_only=True, required=False)
accuracy_radius = serializers.IntegerField(allow_null=True, read_only=True, required=False)
time_zone = serializers.CharField(max_length=100, allow_null=True, allow_blank=True, read_only=True, required=False)
postal_code = serializers.CharField(max_length=20, allow_null=True, allow_blank=True, read_only=True, required=False)
class ReverseGeolocationSerializer(serializers.Serializer):
"""Serializer for reverse geolocation request query parameters"""
lat = serializers.FloatField(
required=True,
min_value=-90.0,
max_value=90.0,
help_text="Latitude coordinate (-90 to 90)"
)
lon = serializers.FloatField(
required=True,
min_value=-180.0,
max_value=180.0,
help_text="Longitude coordinate (-180 to 180)"
)
class ReverseGeolocationResponseSerializer(serializers.Serializer):
"""Serializer for reverse geolocation response"""
latitude = serializers.FloatField(read_only=True)
longitude = serializers.FloatField(read_only=True)
city = serializers.CharField(max_length=100, allow_null=True, read_only=True)
country = serializers.CharField(max_length=100, allow_null=True, read_only=True)
country_code = serializers.CharField(max_length=10, allow_null=True, read_only=True)
accuracy_radius = serializers.IntegerField(allow_null=True, read_only=True, required=False)
time_zone = serializers.CharField(max_length=100, allow_null=True, allow_blank=True, read_only=True, required=False)
postal_code = serializers.CharField(max_length=20, allow_null=True, allow_blank=True, read_only=True, required=False)

10
apps/geolocation_package/utils/__init__.py

@ -0,0 +1,10 @@
from .city_detection_ip import (
get_location_by_coordinates,
get_location_by_ip,
SPECIAL_COORDINATES,
CITY_DB_PATH
)
from .geo import (
get_country_city,
get_country_city_from_point
)

451
apps/geolocation_package/utils/city_detection_ip.py

@ -0,0 +1,451 @@
import os
import time
import geoip2.database
from pathlib import Path
from django.db import connection
from django.db.models import Q
from django.core.cache import cache
from django.db import transaction
import logging
from apps.geolocation_package.models.geoNames import GeoNamesCity
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings.develop')
from django.core.wsgi import get_wsgi_application
application = get_wsgi_application()
from apps.account.models import LoginHistory
# Configure logging
logger = logging.getLogger(__name__)
from config.settings.base import settings
# GeoLite2 database paths
CITY_DB_PATH = Path(settings.BASE_DIR) / "apps" / "geolocation_package" / "data" / "GeoLite2-City.mmdb"
# Special coordinates
SPECIAL_COORDINATES = [
(32.616565, 44.03462),
(51.5287718, -0.2416802),
(40.3947021, 49.78492),
(55.751199, 37.614706),
(48.8589466, 2.2769956),
(40.4381311, -3.8196194),
(-6.2295712, 106.759478),
(33.6158004, 72.8059198)
]
def get_location_by_coordinates_optimized(lat, lon):
"""
Optimized version with special coordinates handling.
Handles special coordinates correctly before geo lookup.
"""
try:
# Quick validation
if not lat or not lon:
return None
lat, lon = float(lat), float(lon)
# Check if coordinates are in special list - should use IP detection instead
for special_lat, special_lon in SPECIAL_COORDINATES:
if abs(lat - special_lat) < 0.001 and abs(lon - special_lon) < 0.001:
# These coordinates should use IP detection, not geo lookup
# Return None to trigger fallback to original method
logger.debug(f"Special coordinate detected: ({lat}, {lon}) - skipping geo lookup")
return None
# Simple cache key
cache_key = f'geo_{round(lat, 2)}_{round(lon, 2)}'
# Try cache first (no exception handling for speed)
cached_result = cache.get(cache_key)
if cached_result is not None:
return cached_result
# Simple bounding box (larger range for better coverage)
lat_range = 3.0 # ~330km
lon_range = 3.0
lat_min = lat - lat_range
lat_max = lat + lat_range
lon_min = lon - lon_range
lon_max = lon + lon_range
# Query with population weighting to prefer larger cities
with connection.cursor() as cursor:
# # First, let's get debug information about nearby cities
# cursor.execute("""
# WITH bounded_cities AS (
# SELECT name, country_code, latitude, longitude, population
# FROM geonames_city
# WHERE feature_class = 'P'
# AND latitude BETWEEN %s AND %s
# AND longitude BETWEEN %s AND %s
# ),
# distance_calc AS (
# SELECT name, country_code, population,
# (6371 * acos(least(1, greatest(-1,
# cos(radians(%s)) * cos(radians(latitude)) *
# cos(radians(longitude) - radians(%s)) +
# sin(radians(%s)) * sin(radians(latitude))
# )))) AS distance
# FROM bounded_cities
# )
# SELECT name, country_code, population, distance
# FROM distance_calc
# WHERE distance <= 100
# ORDER BY distance
# LIMIT 10
# """, [lat_min, lat_max, lon_min, lon_max, lat, lon, lat])
# debug_results = cursor.fetchall()
# if debug_results:
# logger.info(f"🔍 Top 10 nearby cities for coordinates ({lat}, {lon}):")
# for name, cc, pop, dist in debug_results:
# logger.info(f" 📍 {name} ({cc}): population={pop:,}, distance={dist:.2f}km")
# Now get the best city using a weighted approach
# Prefer cities with larger population within reasonable distance
cursor.execute("""
WITH bounded_cities AS (
SELECT name, country_code, latitude, longitude, population
FROM geonames_city
WHERE feature_class = 'P'
AND latitude BETWEEN %s AND %s
AND longitude BETWEEN %s AND %s
AND population IS NOT NULL
AND population > 0
),
distance_calc AS (
SELECT name, country_code, population,
(6371 * acos(least(1, greatest(-1,
cos(radians(%s)) * cos(radians(latitude)) *
cos(radians(longitude) - radians(%s)) +
sin(radians(%s)) * sin(radians(latitude))
)))) AS distance
FROM bounded_cities
),
scored_cities AS (
SELECT name, country_code, distance, population,
-- Score: prefer closer cities, but weight population heavily
-- Cities within 30km: prioritize by population
-- Cities beyond 30km: balance distance and population
CASE
WHEN distance <= 30 THEN population / (distance + 1)
ELSE population / POWER(distance, 2)
END AS score
FROM distance_calc
WHERE distance <= 100
)
SELECT name, country_code
FROM scored_cities
ORDER BY score DESC
LIMIT 1
""", [lat_min, lat_max, lon_min, lon_max, lat, lon, lat])
result = cursor.fetchone()
if result:
name, country_code = result
logger.info(f"✅ Selected city: {name} ({country_code}) for coordinates ({lat}, {lon})")
response = {
'status': 'success',
'city': name,
'countryCode': country_code
}
# Cache for 24 hours
cache.set(cache_key, response, 86400)
return response
else:
logger.warning(f"⚠️ No city found within 100km for coordinates ({lat}, {lon})")
# Cache None for 1 hour
cache.set(cache_key, None, 3600)
return None
except Exception:
# Fallback to original method on any error
return get_location_by_coordinates_original(lat, lon)
def get_location_by_coordinates_original(lat, lon):
"""Original implementation as fallback"""
try:
with connection.cursor() as cursor:
cursor.execute("""
WITH distance_calc AS (
SELECT name, country_code, latitude, longitude,
(6371 * acos(least(1, greatest(-1, cos(radians(%s)) * cos(radians(latitude)) *
cos(radians(longitude) - radians(%s)) +
sin(radians(%s)) * sin(radians(latitude)))))) AS distance
FROM geonames_city
WHERE feature_class = 'P'
)
SELECT name, country_code
FROM distance_calc
WHERE distance <= 300
ORDER BY distance
LIMIT 1
""", [lat, lon, lat])
result = cursor.fetchone()
if result:
name, country_code = result
logger.info(f"🔄 Fallback method selected city: {name} ({country_code}) for coordinates ({lat}, {lon})")
return {
'status': 'success',
'city': name,
'countryCode': country_code
}
return None
except Exception as e:
logger.error(f"❌ Error in fallback method for coordinates ({lat}, {lon}): {str(e)}")
return None
def get_location_by_coordinates(lat, lon):
"""
Main function with smart fallback strategy.
Try optimized first, fallback to original if needed.
"""
# Try optimized version first
result = get_location_by_coordinates_optimized(lat, lon)
# If optimized fails, use original as fallback
if result is None:
result = get_location_by_coordinates_original(lat, lon)
return result
def get_location_by_ip(ip):
"""Get location from IP using MaxMind MMDB file directly"""
try:
if not CITY_DB_PATH.exists():
return None
with geoip2.database.Reader(CITY_DB_PATH) as reader:
response = reader.city(ip)
if response and response.country:
# Validate city name - check if it's not a subdivision
city_name = None
if response.city and response.city.name:
subdivision_names = [s.name for s in response.subdivisions] if response.subdivisions else []
if response.city.name not in subdivision_names:
# City name is valid - not a subdivision
city_name = response.city.name
else:
# City name matches a subdivision - this is a region, not a city
logger.warning(f"IP {ip}: City name '{response.city.name}' matches subdivision - treating as region")
city_name = None
return {
'status': 'success',
'countryCode': response.country.iso_code,
'city': city_name
}
return None
except Exception:
return None
def update_login_history_optimized():
"""
Optimized version with batch processing and better error handling.
Processes records in batches to reduce database load and improve performance.
"""
logger.info("Starting optimized login history update...")
# Query for login histories that need updating
special_records = (
LoginHistory.objects
.exclude(location_method="IP_DETECTION")
.exclude(lat__isnull=True)
.exclude(lon__isnull=True)
.filter(lat__in=[lat for lat, _ in SPECIAL_COORDINATES], lon__in=[lon for _, lon in SPECIAL_COORDINATES])
[:1000] # Limit batch size
)
normal_records = (
LoginHistory.objects
.exclude(location_method="IP_DETECTION")
.exclude(lat__isnull=True)
.exclude(lon__isnull=True)
.exclude(lat__in=[lat for lat, _ in SPECIAL_COORDINATES], lon__in=[lon for _, lon in SPECIAL_COORDINATES])
[:1000] # Limit batch size
)
# Process special coordinates records (with IP) in batches
special_updates = []
for login in special_records:
try:
location_data = get_location_by_ip(login.ip)
if location_data and location_data['status'] == 'success':
login.country = location_data['countryCode']
login.city = location_data['city']
login.location_method = 'IP_DETECTION'
special_updates.append(login)
# Batch update every 50 records
if len(special_updates) >= 50:
with transaction.atomic():
LoginHistory.objects.bulk_update(
special_updates,
['country', 'city', 'location_method']
)
logger.info(f"Updated {len(special_updates)} special coordinate records")
special_updates = []
except Exception as e:
logger.error(f"Error processing special record {login.id}: {e}")
continue
# Final batch update for remaining special records
if special_updates:
with transaction.atomic():
LoginHistory.objects.bulk_update(
special_updates,
['country', 'city', 'location_method']
)
logger.info(f"Updated final {len(special_updates)} special coordinate records")
# Process normal coordinates records (with GeoNames) in batches
normal_updates = []
processed_normal = 0
for login in normal_records:
try:
location_data = get_location_by_coordinates(login.lat, login.lon)
if location_data and location_data['status'] == 'success':
login.country = location_data['countryCode']
login.city = location_data['city']
login.location_method = 'COORDINATES'
normal_updates.append(login)
processed_normal += 1
# Batch update every 20 records (smaller batch for geo queries)
if len(normal_updates) >= 20:
with transaction.atomic():
LoginHistory.objects.bulk_update(
normal_updates,
['country', 'city', 'location_method']
)
logger.info(f"Updated {len(normal_updates)} normal coordinate records")
normal_updates = []
except Exception as e:
logger.error(f"Error processing normal record {login.id}: {e}")
continue
# Final batch update for remaining normal records
if normal_updates:
with transaction.atomic():
LoginHistory.objects.bulk_update(
normal_updates,
['country', 'city', 'location_method']
)
logger.info(f"Updated final {len(normal_updates)} normal coordinate records")
logger.info(f"Completed login history update. Processed {processed_normal} normal records.")
def update_login_history():
"""Backward compatibility wrapper"""
return update_login_history_optimized()
def update_location_history_records_optimized():
"""
Optimized version with batch processing and progress tracking.
Updates location history records with city and country information using GeoNames database.
Only processes records that have coordinates but no city/country information.
"""
from apps.account.models import LocationHistory
logger.info("Starting optimized location history update...")
# Find records that need updating (limit to manageable batch size)
records = LocationHistory.objects.filter(
Q(city__isnull=True) | Q(city='') | Q(country__isnull=True) | Q(country=''),
lat__isnull=False,
lon__isnull=False
)[:1000] # Process in batches of 1000
total_records = records.count()
logger.info(f"Found {total_records} location history records to update")
if total_records == 0:
logger.info("No records to update")
return
updated_count = 0
batch_updates = []
for i, record in enumerate(records, 1):
try:
# Get location data based on coordinates
location_data = get_location_by_coordinates(record.lat, record.lon)
if location_data and location_data['status'] == 'success':
record.city = location_data['city']
record.country = location_data['countryCode']
batch_updates.append(record)
updated_count += 1
# Progress logging every 50 records
if i % 50 == 0:
logger.info(f"Processed {i}/{total_records} records ({updated_count} updated)")
# Batch update every 20 records
if len(batch_updates) >= 20:
with transaction.atomic():
LocationHistory.objects.bulk_update(
batch_updates,
['city', 'country']
)
logger.info(f"Bulk updated {len(batch_updates)} location history records")
batch_updates = []
except Exception as e:
logger.error(f"Error processing location history record {record.id}: {e}")
continue
# Final batch update for remaining records
if batch_updates:
with transaction.atomic():
LocationHistory.objects.bulk_update(
batch_updates,
['city', 'country']
)
logger.info(f"Final bulk update of {len(batch_updates)} location history records")
logger.info(f"Completed location history update. Updated {updated_count}/{total_records} records.")
def update_location_history_records():
"""Backward compatibility wrapper"""
return update_location_history_records_optimized()
if __name__ == "__main__":
# Configure logging for script execution
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('geo_optimization.log'),
logging.StreamHandler()
]
)
logger.info("Starting optimized geo location processing...")
start_time = time.time()
try:
update_login_history()
update_location_history_records()
total_time = time.time() - start_time
logger.info(f"Completed all geo location processing in {total_time:.2f} seconds")
except Exception as e:
logger.error(f"Error in main execution: {e}")
raise

257
apps/geolocation_package/utils/geo.py

@ -0,0 +1,257 @@
import logging
import random
import requests
import time
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
user_agents = [
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Firefox/89.0',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Edge/91.0.864.54',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.1.1 Safari/605.1.15',
'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4430.212 Safari/537.36',
'Mozilla/5.0 (iPhone; CPU iPhone OS 14_6 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.0 Mobile/15E148 Safari/604.1',
'Mozilla/5.0 (iPad; CPU OS 14_6 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.0 Mobile/15E148 Safari/604.1',
'Mozilla/5.0 (Android 11; Mobile; rv:89.0) Gecko/89.0 Firefox/89.0',
'Mozilla/5.0 (Linux; Android 10; SM-G975F) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/89.0.4389.105 Mobile Safari/537.36',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/88.0.4324.104 Safari/537.36'
]
def get_requests_session():
"""Create a requests session with retry logic."""
# Disable urllib3 warnings
import urllib3
urllib3.disable_warnings()
# Disable requests warnings
import requests.packages.urllib3
requests.packages.urllib3.disable_warnings()
session = requests.Session()
# Configure retry strategy without logging
class SilentRetry(Retry):
def increment(self, *args, **kwargs):
# Override to prevent logging
return super().increment(*args, **kwargs)
retry_strategy = SilentRetry(
total=3, # Maximum number of retries
backoff_factor=0.5, # Backoff factor for retries
status_forcelist=[429, 500, 502, 503, 504], # HTTP status codes to retry on
allowed_methods=["GET"] # Only retry on GET requests
)
# Mount the adapter to the session
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("http://", adapter)
session.mount("https://", adapter)
return session
def get_country_city_from_nominatim(lat, lon, headers) -> tuple:
"""Get country and city from OpenStreetMap Nominatim API."""
try:
session = get_requests_session()
resp = session.get(
f"https://nominatim.openstreetmap.org/reverse?lat={lat}&lon={lon}&format=json&addressdetails=1",
headers=headers,
timeout=10
)
if resp.status_code == 200:
address = resp.json().get('address', {})
country = address.get('country_code', '')
# Some responses use 'city', others might use 'town' or 'village'
city = address.get('city', address.get('town', address.get('village', '')))
return country, city
else:
# Silently fail without logging
return '', ''
except Exception:
# Silently fail without logging
return '', ''
def get_country_city_from_bigdatacloud(lat, lon, headers) -> tuple:
"""Get country and city from BigDataCloud API."""
try:
session = get_requests_session()
resp = session.get(
f"https://api.bigdatacloud.net/data/reverse-geocode-client?latitude={lat}&longitude={lon}&localityLanguage=en",
headers=headers,
timeout=10
)
if resp.status_code == 200:
data = resp.json()
country = data.get('countryCode', '')
city = data.get('city', '')
return country, city
else:
return '', ''
except Exception:
# Silently fail without logging
return '', ''
def get_country_city_from_geocode_maps(lat, lon, headers) -> tuple:
"""Get country and city from geocode.maps.co API."""
try:
session = get_requests_session()
resp = session.get(
f"https://geocode.maps.co/reverse?lat={lat}&lon={lon}",
headers=headers,
timeout=10
)
if resp.status_code == 200:
data = resp.json()
address = data.get('address', {})
country = address.get('country_code', '')
city = address.get('city', address.get('town', address.get('village', '')))
return country, city
else:
return '', ''
except Exception:
# Silently fail without logging
return '', ''
def get_country_city_from_ipapi_coordinates(lat, lon, headers) -> tuple:
"""Get country and city from ip-api.com using coordinates."""
try:
# This is a fallback that uses IP geolocation based on server IP
# Not as accurate for the specific coordinates but better than nothing
session = get_requests_session()
resp = session.get(
"http://ip-api.com/json/?fields=status,countryCode,city",
headers=headers,
timeout=10
)
if resp.status_code == 200:
data = resp.json()
if data.get('status') == 'success':
country = data.get('countryCode', '')
city = data.get('city', '')
# Removed info logging
return country, city
return '', ''
except Exception:
# Silently fail without logging
return '', ''
def get_country_city_from_point(lat, lon) -> tuple:
"""Get country and city from coordinates using multiple fallback services."""
headers = {'User-Agent': random.choice(user_agents)}
# List of geocoding functions to try
geocoding_functions = [
get_country_city_from_nominatim,
get_country_city_from_bigdatacloud,
get_country_city_from_geocode_maps,
get_country_city_from_ipapi_coordinates, # Last resort fallback
]
# Try each geocoding service until we get a result
for func in geocoding_functions:
country, city = func(lat, lon, headers)
if country and city:
return country, city
# If all services fail, silently return empty strings without logging
return '', ''
def get_country_city_from_ip(ip, headers) -> tuple:
"""Get country and city from apl.lplocation.net API."""
try:
session = get_requests_session()
resp = session.get(
f"https://apl.lplocation.net/?ip={ip}",
headers=headers,
timeout=5
)
if resp.status_code == 200:
data = resp.json()
country = data.get('country_code', '')
city = data.get('city', '')
return country, city
else:
return '', ''
except Exception:
# Silently fail without logging
return '', ''
def get_country_city_from_ip_api(ip, headers) -> tuple:
"""Get country and city from ip-api.com API."""
try:
session = get_requests_session()
resp = session.get(
f"http://ip-api.com/json/{ip}?fields=status,countryCode,city",
headers=headers,
timeout=5
)
if resp.status_code == 200:
data = resp.json()
if data.get('status') == 'success':
country = data.get('countryCode', '')
city = data.get('city', '')
return country, city
else:
# Silently fail without logging
return '', ''
else:
# Silently fail without logging
return '', ''
except Exception:
# Silently fail without logging
return '', ''
def get_country_city_from_ip2location(ip: str, headers) -> tuple:
"""Get country and city from ip2location.io API."""
try:
session = get_requests_session()
resp = session.get(
f"https://api.ip2location.io/?key=A9DE25AC3ADF5255693F8BFAFB23A902&ip={ip}&format=json",
headers=headers,
timeout=5
)
if resp.status_code == 200:
data = resp.json()
country = data.get('country_code', '')
city = data.get('city_name', '')
return country, city
else:
# Silently fail without logging
return '', ''
except Exception:
# Silently fail without logging
return '', ''
def get_country_city(ip: str) -> tuple:
"""Get country and city from IP using multiple fallback services."""
headers = {'User-Agent': random.choice(user_agents)}
# List of IP-based geocoding functions to try in order of preference
functions = [
get_country_city_from_ip_api, # Most reliable based on common usage
get_country_city_from_ip2location, # Has API key, likely more reliable
get_country_city_from_ip, # Third option
]
# Try each service until we get a result
for i, func in enumerate(functions):
try:
country, city = func(ip, headers)
if country and city:
# No logging for fallbacks
return country, city
# Small delay between API calls to avoid rate limiting
if i < len(functions) - 1:
time.sleep(0.2)
except Exception:
# Silently continue to the next service without logging
continue
# If all services fail, silently return empty strings without logging
return '', ''

2
apps/geolocation_package/views/__init__.py

@ -0,0 +1,2 @@
from .geolocation import IPGeolocationAPIView, ReverseGeolocationAPIView
from .region_info import RegionInfoView

278
apps/geolocation_package/views/geolocation.py

@ -0,0 +1,278 @@
import logging
from pathlib import Path
import geoip2.database
import geoip2.errors
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import status
from drf_yasg.utils import swagger_auto_schema
from drf_yasg import openapi
from apps.geolocation_package.serializers import IPGeolocationSerializer, ReverseGeolocationSerializer, ReverseGeolocationResponseSerializer
from city_detection_ip import get_location_by_coordinates
logger = logging.getLogger(__name__)
# GeoLite2 database path
CITY_DB_PATH = Path("utils/country_city_db/GeoLite2-City.mmdb")
class IPGeolocationAPIView(APIView):
"""
API endpoint to get geolocation information from client's IP address
Returns: country, city, latitude, longitude, timezone, and other location data
"""
permission_classes = []
@swagger_auto_schema(
operation_description="Get geolocation information based on the client's IP address",
responses={
200: openapi.Response(
description="Geolocation information",
schema=IPGeolocationSerializer()
),
404: openapi.Response(
description="IP address not found in database or database not available"
),
500: openapi.Response(
description="Internal server error"
)
},
tags=['account']
)
def get(self, request):
"""Get geolocation info from request IP"""
ip = self.get_client_ip(request)
if not ip:
return Response(
{'error': 'Could not determine client IP address'},
status=status.HTTP_400_BAD_REQUEST
)
# Log the detected IP for debugging
logger.info(f"Detecting location for IP: {ip}")
location_data = self.get_location_from_ip(ip)
if not location_data:
return Response(
{
'error': 'Could not find location data for this IP address',
'ip': ip
},
status=status.HTTP_404_NOT_FOUND
)
# Return location data directly
# Serializer with all read_only fields doesn't work with data parameter
return Response(location_data, status=status.HTTP_200_OK)
def get_client_ip(self, request):
"""Extract client IP from request"""
x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
if x_forwarded_for:
ip = x_forwarded_for.split(',')[0].strip()
else:
ip = request.META.get('REMOTE_ADDR')
return ip
def get_location_from_ip(self, ip):
"""Get location data from IP using GeoIP2 database"""
try:
# Skip local/private IPs
if ip in ['127.0.0.1', 'localhost'] or ip.startswith('192.168.') or ip.startswith('10.'):
logger.warning(f"Skipping local/private IP: {ip}")
# Return empty data instead of None to avoid 404
return {
'ip': ip,
'city': None,
'country': None,
'country_code': None,
'latitude': None,
'longitude': None,
'accuracy_radius': None,
'time_zone': None,
'postal_code': None,
}
if not CITY_DB_PATH.exists():
logger.error(f"GeoIP2 database not found at {CITY_DB_PATH}")
return None
with geoip2.database.Reader(CITY_DB_PATH) as reader:
response = reader.city(ip)
# Extract city name with validation
city_name = None
if response.city and response.city.name:
# Check if city name is actually a subdivision (region)
# This is a known issue in GeoIP2 where subdivision names appear as city names
subdivision_names = [s.name for s in response.subdivisions] if response.subdivisions else []
if response.city.name not in subdivision_names:
# City name is valid - not a subdivision
city_name = response.city.name
else:
# City name matches a subdivision - this is a region, not a city
logger.warning(f"IP {ip}: City name '{response.city.name}' matches subdivision - treating as region")
city_name = None # Don't return region as city
location_data = {
'ip': ip,
'city': city_name,
'country': response.country.name if response.country else None,
'country_code': response.country.iso_code if response.country else None,
'latitude': response.location.latitude if response.location else None,
'longitude': response.location.longitude if response.location else None,
'accuracy_radius': response.location.accuracy_radius if response.location else None,
'time_zone': response.location.time_zone if response.location else None,
'postal_code': response.postal.code if response.postal else None,
}
logger.info(f"Successfully found location for IP {ip}: {location_data.get('city')}, {location_data.get('country')}")
return location_data
except geoip2.errors.AddressNotFoundError:
logger.warning(f"IP address {ip} not found in GeoIP2 database")
return None
except Exception as e:
logger.error(f"Error getting location from IP {ip}: {str(e)}")
return None
class ReverseGeolocationAPIView(APIView):
"""
API endpoint to get location information from geographic coordinates
Returns: city, country, country_code based on latitude and longitude
"""
permission_classes = []
def validate_city_name_from_coordinates(self, lat, lon, city_name):
"""
Validate that the city name is not actually a subdivision (region).
Uses keyword-based heuristic to detect subdivision names.
Args:
lat: Latitude coordinate
lon: Longitude coordinate
city_name: City name to validate
Returns:
Validated city name or None if it's a subdivision
"""
if not city_name:
return None
try:
# Simple heuristic: if city name contains common subdivision keywords
# in various languages, it might be a subdivision
subdivision_keywords = [
'Province', 'Region', 'Oblast', 'Governorate',
'District', 'County', 'State', 'Territory',
'استان', 'منطقه', 'ولایت', 'محافظه'
]
for keyword in subdivision_keywords:
if keyword.lower() in city_name.lower():
logger.warning(
f"⚠️ City name '{city_name}' at ({lat}, {lon}) "
f"contains subdivision keyword '{keyword}' - treating as region (returning None)"
)
return None
logger.debug(f"✅ City name '{city_name}' validated for ({lat}, {lon})")
return city_name
except Exception as e:
logger.error(f"❌ Error validating city name for coordinates ({lat}, {lon}): {str(e)}")
return city_name # Return as-is on error
@swagger_auto_schema(
operation_description="Get location information (city, country) based on geographic coordinates using reverse geocoding",
manual_parameters=[
openapi.Parameter(
'lat',
openapi.IN_QUERY,
description="Latitude coordinate (-90 to 90)",
type=openapi.TYPE_NUMBER,
required=True
),
openapi.Parameter(
'lon',
openapi.IN_QUERY,
description="Longitude coordinate (-180 to 180)",
type=openapi.TYPE_NUMBER,
required=True
),
],
responses={
200: openapi.Response(
description="Location information",
schema=ReverseGeolocationResponseSerializer()
),
400: openapi.Response(
description="Invalid or missing coordinates"
),
404: openapi.Response(
description="No location found for the given coordinates"
),
500: openapi.Response(
description="Internal server error"
)
},
tags=['account']
)
def get(self, request):
"""Get location info from coordinates"""
# Validate query parameters
serializer = ReverseGeolocationSerializer(data=request.query_params)
if not serializer.is_valid():
return Response(
{
'error': 'Invalid coordinates',
'details': serializer.errors
},
status=status.HTTP_400_BAD_REQUEST
)
lat = serializer.validated_data['lat']
lon = serializer.validated_data['lon']
# Log the coordinates for debugging
logger.info(f"Reverse geocoding for coordinates: ({lat}, {lon})")
# Get location data using the existing function from city_detection_ip.py
location_data = get_location_by_coordinates(lat, lon)
if not location_data or location_data.get('status') != 'success':
return Response(
{
'error': 'Could not find location data for these coordinates',
'latitude': lat,
'longitude': lon
},
status=status.HTTP_404_NOT_FOUND
)
# Validate city name to ensure it's not a subdivision (region)
city_name = location_data.get('city')
validated_city = self.validate_city_name_from_coordinates(lat, lon, city_name)
# Format response
response_data = {
'latitude': lat,
'longitude': lon,
'city': validated_city,
'country': None, # GeoNames only returns country_code
'country_code': location_data.get('countryCode'),
'accuracy_radius': None,
'time_zone': None,
'postal_code': None,
}
logger.info(f"Successfully found location for coordinates ({lat}, {lon}): {response_data.get('city')}, {response_data.get('country_code')}")
return Response(response_data, status=status.HTTP_200_OK)

186
apps/geolocation_package/views/region_info.py

@ -0,0 +1,186 @@
from rest_framework.generics import GenericAPIView
from rest_framework.response import Response
import logging
import re
from pathlib import Path
import geoip2.database
import geoip2.errors
logger = logging.getLogger(__name__)
# GeoLite2 database path
CITY_DB_PATH = Path("utils/country_city_db/GeoLite2-City.mmdb")
def detect_browser_from_user_agent(user_agent):
"""
Detect browser name from User-Agent string.
Args:
user_agent (str): The User-Agent header value
Returns:
str or None: Browser name if detected, None if not a browser or detection fails
"""
if not user_agent:
return None
try:
user_agent = user_agent.lower()
# Check for Flutter/Dart app (return None for non-browser requests)
if any(keyword in user_agent for keyword in ['dart:io', 'flutter', 'dart/']):
return None
# Check for mobile apps that might not be browsers
if any(keyword in user_agent for keyword in ['habibapp', 'mobile app']):
return None
# Browser detection patterns (order matters - more specific first)
browser_patterns = [
(r'edg/', 'Edge'), # Microsoft Edge (Chromium-based)
(r'edge/', 'Edge'), # Microsoft Edge (Legacy)
(r'opr/', 'Opera'), # Opera
(r'opera/', 'Opera'), # Opera
(r'chrome/', 'Chrome'), # Google Chrome
(r'chromium/', 'Chromium'), # Chromium
(r'firefox/', 'Firefox'), # Mozilla Firefox
(r'fxios/', 'Firefox'), # Firefox for iOS
(r'safari/', 'Safari'), # Safari (check after Chrome/Edge as they also contain Safari)
(r'version/.*safari', 'Safari'), # Safari with version
]
# Check each pattern
for pattern, browser_name in browser_patterns:
if re.search(pattern, user_agent):
# Additional check for Safari to avoid false positives
if browser_name == 'Safari':
# Make sure it's not Chrome, Edge, or other browsers that include Safari in UA
if not any(other in user_agent for other in ['chrome', 'edg', 'opr', 'opera']):
return browser_name
else:
return browser_name
# If no specific browser detected but contains Mozilla, it might be an unknown browser
if 'mozilla' in user_agent and any(keyword in user_agent for keyword in ['gecko', 'webkit']):
return 'Unknown Browser'
return None
except Exception as e:
# Log the error but don't let it break the API
logger.warning(f"Error detecting browser from user agent: {e}")
return None
class RegionInfoView(GenericAPIView):
def get(self, request, *args, **kwargs):
# Get browser information safely
browser = None
try:
user_agent = request.META.get('HTTP_USER_AGENT', '')
browser = detect_browser_from_user_agent(user_agent)
except Exception as e:
# Log the error but continue with the API response
logger.warning(f"Error detecting browser in RegionInfoView: {e}")
browser = None
# Get IP address
ip = self.get_client_ip(request)
# Get geolocation data from GeoIP2 database
geo_data = self.get_location_from_ip(ip)
region_info = {
'ip': request.META.get('HTTP_CF_CONNECTING_IP') or ip,
'country': request.META.get('HTTP_CF_IPCOUNTRY'),
'region': request.META.get('HTTP_CF_REGION'),
'region_code': request.META.get('HTTP_CF_REGION_CODE'),
'city': request.META.get('HTTP_CF_CITY'),
'timezone': request.META.get('HTTP_CF_TIMEZONE'),
'browser': browser,
}
# Add geolocation data if available
if geo_data:
region_info.update({
'country_code': geo_data.get('country_code'),
'latitude': geo_data.get('latitude'),
'longitude': geo_data.get('longitude'),
'accuracy_radius': geo_data.get('accuracy_radius'),
'time_zone': geo_data.get('time_zone'),
'postal_code': geo_data.get('postal_code'),
})
return Response(region_info)
def get_client_ip(self, request):
"""Extract client IP from request"""
x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
if x_forwarded_for:
ip = x_forwarded_for.split(',')[0].strip()
else:
ip = request.META.get('REMOTE_ADDR')
return ip
def get_location_from_ip(self, ip):
"""Get location data from IP using GeoIP2 database"""
try:
# Skip local/private IPs
if ip in ['127.0.0.1', 'localhost'] or ip.startswith('192.168.') or ip.startswith('10.'):
logger.warning(f"Skipping local/private IP: {ip}")
return {
'ip': ip,
'city': None,
'country': None,
'country_code': None,
'latitude': None,
'longitude': None,
'accuracy_radius': None,
'time_zone': None,
'postal_code': None,
}
if not CITY_DB_PATH.exists():
logger.error(f"GeoIP2 database not found at {CITY_DB_PATH}")
return None
with geoip2.database.Reader(CITY_DB_PATH) as reader:
response = reader.city(ip)
# Extract city name with validation
city_name = None
if response.city and response.city.name:
# Check if city name is actually a subdivision (region)
# This is a known issue in GeoIP2 where subdivision names appear as city names
subdivision_names = [s.name for s in response.subdivisions] if response.subdivisions else []
if response.city.name not in subdivision_names:
# City name is valid - not a subdivision
city_name = response.city.name
else:
# City name matches a subdivision - this is a region, not a city
logger.warning(f"IP {ip}: City name '{response.city.name}' matches subdivision - treating as region")
city_name = None # Don't return region as city
location_data = {
'ip': ip,
'city': city_name,
'country': response.country.name if response.country else None,
'country_code': response.country.iso_code if response.country else None,
'latitude': response.location.latitude if response.location else None,
'longitude': response.location.longitude if response.location else None,
'accuracy_radius': response.location.accuracy_radius if response.location else None,
'time_zone': response.location.time_zone if response.location else None,
'postal_code': response.postal.code if response.postal else None,
}
logger.info(f"Successfully found location for IP {ip}: {location_data.get('city')}, {location_data.get('country')}")
return location_data
except geoip2.errors.AddressNotFoundError:
logger.warning(f"IP address {ip} not found in GeoIP2 database")
return None
except Exception as e:
logger.error(f"Error getting location from IP {ip}: {str(e)}")
return None

1
config/settings/base.py

@ -58,6 +58,7 @@ LOCAL_APPS = [
'apps.dobodbi_calendar.apps.DobodbiCalendarConfig', 'apps.dobodbi_calendar.apps.DobodbiCalendarConfig',
'apps.blog.apps.BlogConfig', 'apps.blog.apps.BlogConfig',
'dynamic_preferences', 'dynamic_preferences',
'apps.geolocation_package',
] ]

Loading…
Cancel
Save