Add real IP + geo location to Telegram notifications
This commit is contained in:
+42
@@ -65,6 +65,32 @@ logging.basicConfig(
|
|||||||
logger = logging.getLogger("portal")
|
logger = logging.getLogger("portal")
|
||||||
|
|
||||||
templates = Jinja2Templates(directory="templates")
|
templates = Jinja2Templates(directory="templates")
|
||||||
|
|
||||||
|
def _get_real_ip(request) -> str:
|
||||||
|
"""Return real client IP, accounting for NPM → app proxy chain."""
|
||||||
|
forwarded_for = request.headers.get("x-forwarded-for", "")
|
||||||
|
if forwarded_for:
|
||||||
|
# X-Forwarded-For: client, proxy1, proxy2 — take leftmost (real client)
|
||||||
|
return forwarded_for.split(",")[0].strip()
|
||||||
|
return request.client.host if request.client else "unknown"
|
||||||
|
|
||||||
|
|
||||||
|
def _get_geo(ip: str) -> str:
|
||||||
|
"""Lookup city/country for IP via ip-api.com. Returns formatted string or empty."""
|
||||||
|
try:
|
||||||
|
if ip in ("unknown", "127.0.0.1", "::1") or ip.startswith("10.") or ip.startswith("192.168."):
|
||||||
|
return ""
|
||||||
|
url = f"http://ip-api.com/json/{ip}?lang=ru&fields=status,country,regionName,city,query"
|
||||||
|
req = _urllib_request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
|
||||||
|
with _urllib_request.urlopen(req, timeout=5) as resp:
|
||||||
|
data = _json.loads(resp.read())
|
||||||
|
if data.get("status") == "success":
|
||||||
|
parts = [data.get("country", ""), data.get("regionName", ""), data.get("city", "")]
|
||||||
|
return ", ".join(p for p in parts if p)
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
return ""
|
||||||
|
|
||||||
app = FastAPI(title="МОНТ - инфрастуктурный полигон")
|
app = FastAPI(title="МОНТ - инфрастуктурный полигон")
|
||||||
app.mount("/static", StaticFiles(directory="static"), name="static")
|
app.mount("/static", StaticFiles(directory="static"), name="static")
|
||||||
|
|
||||||
@@ -466,6 +492,14 @@ async def request_access(request: Request, db: Session = Depends(get_db)):
|
|||||||
f"{products_text}"
|
f"{products_text}"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
ip = _get_real_ip(request)
|
||||||
|
geo = _get_geo(ip)
|
||||||
|
geo_text = ""
|
||||||
|
if geo:
|
||||||
|
geo_text += "\n📍 *Местоположение:* " + geo
|
||||||
|
geo_text += "\n🖥 *IP:* " + ip
|
||||||
|
text += geo_text
|
||||||
|
|
||||||
if not TELEGRAM_BOT_TOKEN or not TELEGRAM_CHAT_ID:
|
if not TELEGRAM_BOT_TOKEN or not TELEGRAM_CHAT_ID:
|
||||||
log_event("telegram_not_configured")
|
log_event("telegram_not_configured")
|
||||||
return {"ok": True}
|
return {"ok": True}
|
||||||
@@ -519,6 +553,14 @@ async def contact_ruslan(request: Request):
|
|||||||
f"💬 *Сообщение:*\n{text}"
|
f"💬 *Сообщение:*\n{text}"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
ip = _get_real_ip(request)
|
||||||
|
geo = _get_geo(ip)
|
||||||
|
geo_text = ""
|
||||||
|
if geo:
|
||||||
|
geo_text += f"\n📍 *Местоположение:* {geo}"
|
||||||
|
geo_text += f"\n🖥 *IP:* {ip}"
|
||||||
|
msg += geo_text
|
||||||
|
|
||||||
if not TELEGRAM_BOT_TOKEN or not TELEGRAM_CHAT_ID:
|
if not TELEGRAM_BOT_TOKEN or not TELEGRAM_CHAT_ID:
|
||||||
log_event("telegram_not_configured")
|
log_event("telegram_not_configured")
|
||||||
return {"ok": True}
|
return {"ok": True}
|
||||||
|
|||||||
Reference in New Issue
Block a user