Django/utils/link_sender_meta.py

# utils/link_sender_meta.py
"""商家订单链接页：采集实际发消息者（浏览器/网络）信息，不涉及客服 JWT 或账号隐私。"""
import logging
import re

import requests
from django.utils import timezone

from config.views import IPUtils

logger = logging.getLogger(__name__)

_TRACKING_HEADER_KEYS = (
    'HTTP_ACCEPT',
    'HTTP_ACCEPT_ENCODING',
    'HTTP_ACCEPT_LANGUAGE',
    'HTTP_REFERER',
    'HTTP_ORIGIN',
    'HTTP_HOST',
    'HTTP_USER_AGENT',
    'HTTP_X_FORWARDED_FOR',
    'HTTP_X_REAL_IP',
    'HTTP_VIA',
    'HTTP_CF_CONNECTING_IP',
    'HTTP_TRUE_CLIENT_IP',
    'HTTP_X_CLIENT_IP',
    'HTTP_SEC_CH_UA',
    'HTTP_SEC_CH_UA_MOBILE',
    'HTTP_SEC_CH_UA_PLATFORM',
    'HTTP_SEC_FETCH_SITE',
    'HTTP_SEC_FETCH_MODE',
    'HTTP_SEC_FETCH_DEST',
    'HTTP_DNT',
    'HTTP_CONNECTION',
)


def _lookup_ip_geo(ip):
    """通过 ip-api.com 查询 IP 归属（免费，无需 key）"""
    if not ip or not IPUtils.validate_ip(ip) or IPUtils.is_private_ip(ip):
        return {'note': '内网或非公网 IP，无法定位'}

    try:
        fields = (
            'status,message,country,regionName,city,district,zip,lat,lon,'
            'timezone,isp,org,as,query,mobile,proxy,hosting'
        )
        resp = requests.get(
            f'http://ip-api.com/json/{ip}?lang=zh-CN&fields={fields}',
            timeout=3,
        )
        if resp.status_code != 200:
            return {}
        data = resp.json()
        if data.get('status') != 'success':
            return {'note': data.get('message') or '定位查询失败'}
        return {
            'country': data.get('country'),
            'region': data.get('regionName'),
            'city': data.get('city'),
            'district': data.get('district'),
            'zip': data.get('zip'),
            'lat': data.get('lat'),
            'lon': data.get('lon'),
            'timezone': data.get('timezone'),
            'isp': data.get('isp'),
            'org': data.get('org'),
            'as': data.get('as'),
            'is_mobile_network': data.get('mobile'),
            'is_proxy': data.get('proxy'),
            'is_hosting': data.get('hosting'),
        }
    except Exception as e:
        logger.warning('IP 归属查询失败: %s', e)
        return {}


def _parse_ip_chain(xff):
    if not xff:
        return []
    return [p.strip() for p in xff.split(',') if p.strip()]


def _build_ip_geo_chain(primary_ip, xff):
    """主 IP + 代理链上最多 2 个公网 IP 的归属"""
    chain = []
    seen = set()
    candidates = [primary_ip] + _parse_ip_chain(xff or '')
    for ip in candidates:
        if not ip or ip in seen:
            continue
        seen.add(ip)
        if not IPUtils.validate_ip(ip):
            continue
        chain.append({'ip': ip, 'geo': _lookup_ip_geo(ip)})
        if len(chain) >= 2:
            break
    return chain


def _parse_user_agent(ua):
    if not ua:
        return {}
    ua = ua[:500]
    info = {'raw': ua}
    if 'MicroMessenger' in ua:
        info['browser'] = '微信内置浏览器'
    elif 'Edg/' in ua:
        info['browser'] = 'Microsoft Edge'
    elif 'Chrome/' in ua and 'Safari/' in ua:
        info['browser'] = 'Chrome'
    elif 'Firefox/' in ua:
        info['browser'] = 'Firefox'
    elif 'Safari/' in ua:
        info['browser'] = 'Safari'
    else:
        info['browser'] = '未知浏览器'

    if 'Windows NT 10' in ua:
        info['os'] = 'Windows 10/11'
    elif 'Windows' in ua:
        info['os'] = 'Windows'
    elif 'Mac OS X' in ua or 'Macintosh' in ua:
        info['os'] = 'macOS'
    elif 'Android' in ua:
        m = re.search(r'Android\s+([\d.]+)', ua)
        info['os'] = f"Android {m.group(1)}" if m else 'Android'
    elif 'iPhone' in ua or 'iPad' in ua:
        info['os'] = 'iOS'
    elif 'Linux' in ua:
        info['os'] = 'Linux'
    else:
        info['os'] = '未知系统'

    info['device_type'] = 'mobile' if ('Mobile' in ua or 'Android' in ua or 'iPhone' in ua) else 'desktop'
    return info


def _collect_tracking_headers(request):
    out = {}
    for key in _TRACKING_HEADER_KEYS:
        val = request.META.get(key)
        if val:
            name = key.replace('HTTP_', '').lower().replace('_', '-')
            out[name] = str(val)[:800]
    return out


def build_link_sender_meta(request, link_token='', client_info=None):
    """
    构建实际发消息者（浏览器访问者）的追踪元数据。
    不读取 JWT / 客服账号，避免泄露客服隐私。
    client_info: 前端 H5 上报的设备指纹与环境信息。
    """
    ip = IPUtils.get_client_ip(request)
    xff = request.META.get('HTTP_X_FORWARDED_FOR', '') or ''
    cf_ip = request.META.get('HTTP_CF_CONNECTING_IP', '') or None
    true_client_ip = request.META.get('HTTP_TRUE_CLIENT_IP', '') or None

    meta = {
        'send_time': timezone.now().strftime('%Y-%m-%d %H:%M:%S'),
        'ip': ip,
        'x_forwarded_for': xff or None,
        'x_real_ip': request.META.get('HTTP_X_REAL_IP', '') or None,
        'remote_addr': request.META.get('REMOTE_ADDR', '') or None,
        'cf_connecting_ip': cf_ip,
        'true_client_ip': true_client_ip,
        'ip_geo_chain': _build_ip_geo_chain(ip, xff),
        'geo': _lookup_ip_geo(ip),
        'user_agent': request.META.get('HTTP_USER_AGENT', ''),
        'browser': _parse_user_agent(request.META.get('HTTP_USER_AGENT', '')),
        'headers': _collect_tracking_headers(request),
        'link_token_prefix': (link_token[:8] + '...') if link_token else None,
    }

    if isinstance(client_info, dict) and client_info:
        meta['device'] = client_info

    return meta