# -*- coding: utf-8 -*-
"""
IP地理位置查詢服務(wù)
支持多種查詢方式和異常處理
"""
import json
import requests
from bs4 import BeautifulSoup
from datetime import datetime
import socket
from typing import Optional, Dict, List
import time
from requests.exceptions import RequestException, Timeout, ConnectionError, HTTPError
import logging
# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class IPinfoService:
    """IP地理位置查詢服務(wù)類"""
    def __init__(self):
        """初始化服務(wù)"""
        self.timeout = 10  # 默認(rèn)超時(shí)時(shí)間
        self.max_retries = 3  # 最大重試次數(shù)
        self.session = requests.Session()
        self.session.headers.update({
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        })
    def get_all_ips(self) -> List[str]:
        """
        獲取本機(jī)所有網(wǎng)絡(luò)接口的IP地址
        Returns:
            IP地址列表
        """
        ips = []
        hostname = socket.gethostname()
        try:
            for addr_info in socket.getaddrinfo(hostname, None):
                family, _, _, _, sockaddr = addr_info
                if family == socket.AF_INET:
                    ips.append(sockaddr[0])
            logger.info(f"獲取到 {len(ips)} 個(gè)本地IP地址")
            return ips
        except Exception as e:
            logger.error(f"獲取本地IP失敗: {e}")
            return []
    def get_local_ip(self) -> str:
        """
        獲取本機(jī)主要IP地址
        Returns:
            本機(jī)IP地址
        """
        try:
            # 創(chuàng)建UDP socket連接到公共DNS服務(wù)器
            s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            s.connect(("8.8.8.8", 80))
            local_ip = s.getsockname()[0]
            s.close()
            logger.info(f"獲取到本機(jī)IP: {local_ip}")
            return local_ip
        except Exception as e:
            logger.error(f"獲取本機(jī)IP失敗: {e}")
            return "127.0.0.1"  # 回退地址
    def get_public_ip(self) -> Optional[str]:
        """
        獲取本機(jī)公網(wǎng)IP地址
        Returns:
            公網(wǎng)IP地址或None
        """
        services = [
            'https://api.ipify.org',
            'https://ifconfig.me/ip',
            'https://icanhazip.com',
            'https://api.ip.sb/ip'
        ]
        for service in services:
            try:
                response = self.session.get(service, timeout=self.timeout)
                response.raise_for_status()
                ip = response.text.strip()
                # 驗(yàn)證IP格式
                if self._validate_ip(ip):
                    logger.info(f"通過 {service} 獲取到公網(wǎng)IP: {ip}")
                    return ip
            except Exception as e:
                logger.debug(f"服務(wù) {service} 獲取失敗: {e}")
                continue
        logger.error("所有公網(wǎng)IP查詢服務(wù)均失敗")
        return None
    def get_city_by_ip_api(self, ip_address: str) -> Optional[str]:
        """
        使用ip-api.com查詢IP地理位置
        Args:
            ip_address: IP地址
        Returns:
            城市名稱或None
        """
        url = f"http://ip-api.com/json/{ip_address}"
        try:
            response = self.session.get(url, timeout=self.timeout)
            response.raise_for_status()
            data = response.json()
            if data.get('status') == 'success':
                city = data.get('city', '未知城市')
                logger.info(f"IP {ip_address} 通過ip-api.com定位到: {city}")
                return city
            else:
                logger.warning(f"IP {ip_address} 查詢失敗: {data.get('message')}")
                return None
        except Exception as e:
            logger.error(f"ip-api.com查詢異常: {e}")
            return None
    def get_city_by_ipinfo(self, ip_address: str) -> Optional[str]:
        """
        使用ipinfo.io查詢IP地理位置
        Args:
            ip_address: IP地址
        Returns:
            城市名稱或None
        """
        url = f"https://ipinfo.io/{ip_address}/json"
        try:
            response = self.session.get(url, timeout=self.timeout)
            response.raise_for_status()
            data = response.json()
            city = data.get('city', '未知城市')
            logger.info(f"IP {ip_address} 通過ipinfo.io定位到: {city}")
            return city
        except Exception as e:
            logger.error(f"ipinfo.io查詢異常: {e}")
            return None
    def get_location_by_baidu(self, ip_address: str, ak: Optional[str] = None) -> Optional[Dict]:
        """
        使用百度地圖API查詢IP地理位置
        Args:
            ip_address: IP地址
            ak: 百度地圖API Key（可選）
        Returns:
            位置信息字典或None
        """
        if not ak:
            logger.warning("未提供百度地圖API Key")
            return None
        url = "https://api.map.baidu.com/location/ip"
        params = {
            'ip': ip_address,
            'ak': ak,
            'coor': 'bd09ll'
        }
        try:
            response = self.session.get(url, params=params, timeout=self.timeout)
            response.raise_for_status()
            data = response.json()
            if data.get('status') == 0:
                content = data.get('content', {})
                address_detail = content.get('address_detail', {})
                location_info = {
                    'ip': data.get('address'),
                    'province': address_detail.get('province'),
                    'city': address_detail.get('city'),
                    'district': address_detail.get('district'),
                    'latitude': content.get('point', {}).get('y'),
                    'longitude': content.get('point', {}).get('x'),
                    'source': '百度地圖API'
                }
                logger.info(f"IP {ip_address} 通過百度地圖API定位到: {location_info.get('city')}")
                return location_info
            else:
                logger.warning(f"百度地圖API查詢失敗: {data.get('message')}")
                return None
        except Exception as e:
            logger.error(f"百度地圖API查詢異常: {e}")
            return None
    def get_location_from_baidu_search(self) -> Optional[Dict]:
        """
        通過爬取百度搜索頁面獲取IP地理位置信息
        Returns:
            包含IP和位置信息的字典或None
        """
        url = 'https://www.baidu.com/s'
        params = {'wd': 'IP', 'ie': 'utf-8'}
        try:
            response = self.session.get(url, params=params, timeout=self.timeout)
            response.encoding = 'utf-8'
            # 使用正則表達(dá)式提取信息
            import re
            ip_pattern = r'本機(jī)IP[：:]\s*([\d.]+)'
            location_pattern = r'來自[：:]\s*([^<\s]+)'
            ip_match = re.search(ip_pattern, response.text)
            location_match = re.search(location_pattern, response.text)
            result = {
                'ip': ip_match.group(1) if ip_match else None,
                'location': location_match.group(1) if location_match else None,
                'source': '百度搜索'
            }
            logger.info(f"通過百度搜索獲取到IP: {result.get('ip')}, 位置: {result.get('location')}")
            return result
        except Exception as e:
            logger.error(f"百度搜索查詢異常: {e}")
            return None
    def compare_services(self, ip_address: Optional[str] = None) -> Dict[str, Dict]:
        """
        對(duì)比不同IP查詢服務(wù)的結(jié)果
        Args:
            ip_address: 要查詢的IP地址，如果為None則查詢當(dāng)前公網(wǎng)IP
        Returns:
            各服務(wù)查詢結(jié)果的字典
        """
        if ip_address is None:
            ip_address = self.get_public_ip()
        if not ip_address:
            logger.error("無法獲取IP地址進(jìn)行對(duì)比")
            return {}
        services = {
            'ip-api': self.get_city_by_ip_api,
            'ipinfo': self.get_city_by_ipinfo,
        }
        results = {}
        for name, method in services.items():
            try:
                start_time = time.time()
                result = method(ip_address)
                elapsed = time.time() - start_time
                results[name] = {
                    'city': result,
                    'response_time': f"{elapsed:.3f}s",
                    'status': '成功'
                }
                # 避免請(qǐng)求過快
                time.sleep(0.5)
            except Exception as e:
                results[name] = {
                    'status': '失敗',
                    'error': str(e)
                }
        logger.info("服務(wù)對(duì)比完成")
        return results
    def _validate_ip(self, ip: str) -> bool:
        """
        驗(yàn)證IP地址格式
        Args:
            ip: IP地址字符串
        Returns:
            是否為有效IP地址
        """
        try:
            socket.inet_aton(ip)
            return True
        except socket.error:
            return False
    def get_location_info(self, ip_address: str) -> Dict:
        """
        綜合獲取IP地址的位置信息
        Args:
            ip_address: IP地址
        Returns:
            包含多種信息的位置字典
        """
        info = {
            'ip': ip_address,
            'timestamp': datetime.now().isoformat(),
            'sources': {}
        }
        # 嘗試不同的查詢服務(wù)
        services = [
            ('ip-api', self.get_city_by_ip_api),
            ('ipinfo', self.get_city_by_ipinfo),
        ]
        for name, method in services:
            try:
                result = method(ip_address)
                if result:
                    info['sources'][name] = {
                        'city': result,
                        'status': 'success'
                    }
                else:
                    info['sources'][name] = {
                        'status': 'failed',
                        'message': '查詢失敗'
                    }
            except Exception as e:
                info['sources'][name] = {
                    'status': 'error',
                    'error': str(e)
                }
        # 選擇最可信的結(jié)果
        for name in ['ip-api', 'ipinfo']:
            if name in info['sources'] and info['sources'][name].get('status') == 'success':
                info['best_match'] = info['sources'][name]['city']
                break
        logger.info(f"獲取IP {ip_address} 的位置信息完成")
        return info
    def close(self):
        """關(guān)閉會(huì)話"""
        self.session.close()
        logger.info("IP查詢服務(wù)會(huì)話已關(guān)閉")
def main():
    """主函數(shù)"""
    print("=" * 60)
    print("IP地理位置查詢服務(wù)演示")
    print("=" * 60)
    service = IPinfoService()
    try:
        # 1. 獲取本機(jī)IP信息
        print("\n### 本機(jī)網(wǎng)絡(luò)信息 ###")
        local_ips = service.get_all_ips()
        print(f"所有網(wǎng)絡(luò)接口IP: {local_ips}")
        local_ip = service.get_local_ip()
        print(f"本機(jī)主要IP: {local_ip}")
        # 2. 獲取公網(wǎng)IP
        print("\n### 公網(wǎng)IP信息 ###")
        public_ip = service.get_public_ip()
        if public_ip:
            print(f"本機(jī)公網(wǎng)IP: {public_ip}")
            # 3. 查詢公網(wǎng)IP的地理位置
            print("\n### 公網(wǎng)IP地理位置查詢 ###")
            # 使用不同服務(wù)查詢
            print("1. 使用ip-api.com查詢:")
            city1 = service.get_city_by_ip_api(public_ip)
            print(f"   城市: {city1}")
            print("\n2. 使用ipinfo.io查詢:")
            city2 = service.get_city_by_ipinfo(public_ip)
            print(f"   城市: {city2}")
            # 4. 對(duì)比服務(wù)結(jié)果
            print("\n### 服務(wù)結(jié)果對(duì)比 ###")
            comparison = service.compare_services(public_ip)
            for service_name, result in comparison.items():
                print(f"{service_name}:")
                if result.get('status') == '成功':
                    print(f"  城市: {result.get('city')}")
                    print(f"  響應(yīng)時(shí)間: {result.get('response_time')}")
                else:
                    print(f"  狀態(tài): {result.get('status')}")
                    print(f"  錯(cuò)誤: {result.get('error', '未知錯(cuò)誤')}")
            # 5. 綜合位置信息
            print("\n### 綜合位置信息 ###")
            location_info = service.get_location_info(public_ip)
            print(json.dumps(location_info, indent=2, ensure_ascii=False))
        else:
            print("無法獲取公網(wǎng)IP地址")
        # 6. 查詢指定IP
        print("\n### 查詢指定IP ###")
        test_ips = ["8.8.8.8", "114.114.114.114", "1.1.1.1"]
        for ip in test_ips:
            print(f"\n查詢IP: {ip}")
            city = service.get_city_by_ip_api(ip)
            print(f"城市: {city}")
            time.sleep(1)  # 避免請(qǐng)求過快
    finally:
        service.close()
        print("\n" + "=" * 60)
        print("演示結(jié)束")
        print("=" * 60)
if __name__ == '__main__':
    main()

import requests
import time
from requests.exceptions import HTTPError, ConnectionError, Timeout, RequestException
from typing import Optional, Dict, Any
class RobustIPQuery:
    """帶異常處理和重試機(jī)制的IP查詢服務(wù)"""
    
    def __init__(self, timeout: int = 10, max_retries: int = 3):
        """
        初始化
        
        Args:
            timeout: 超時(shí)時(shí)間（秒）
            max_retries: 最大重試次數(shù)
        """
        self.timeout = timeout
        self.max_retries = max_retries
        self.session = requests.Session()
    
    def call_api_with_retry(
        self,
        url: str,
        method: str = "GET",
        params: Optional[Dict] = None,
        json_data: Optional[Dict] = None,
        headers: Optional[Dict] = None
    ) -> Optional[Dict[Any, Any]]:
        """
        帶異常處理和重試機(jī)制的API調(diào)用函數(shù)
        
        Args:
            url: 接口URL
            method: 請(qǐng)求方法（GET/POST）
            params: GET請(qǐng)求參數(shù)
            json_data: POST請(qǐng)求體
            headers: 請(qǐng)求頭
            
        Returns:
            接口返回的JSON數(shù)據(jù)或None
        """
        # 初始化請(qǐng)求參數(shù)
        params = params or {}
        json_data = json_data or {}
        headers = headers or {"Content-Type": "application/json"}
        
        retries = 0
        retry_delay = 1  # 初始重試間隔
        
        while retries <= self.max_retries:
            try:
                # 發(fā)送請(qǐng)求
                if method.upper() == "GET":
                    response = self.session.get(
                        url,
                        params=params,
                        headers=headers,
                        timeout=self.timeout
                    )
                else:  # POST
                    response = self.session.post(
                        url,
                        json=json_data,
                        headers=headers,
                        timeout=self.timeout
                    )
                
                # 檢查HTTP狀態(tài)碼
                response.raise_for_status()
                
                # 返回JSON數(shù)據(jù)
                return response.json()
                
            except HTTPError as e:
                status_code = e.response.status_code if e.response else None
                
                # 4xx客戶端錯(cuò)誤，不需要重試
                if status_code and 400 <= status_code < 500:
                    print(f"客戶端錯(cuò)誤: {status_code}, 不重試")
                    return None
                
                # 5xx服務(wù)端錯(cuò)誤，可以重試
                print(f"服務(wù)端錯(cuò)誤: {status_code}, 重試 {retries + 1}/{self.max_retries}")
                
            except (ConnectionError, Timeout) as e:
                print(f"網(wǎng)絡(luò)異常: {e}, 重試 {retries + 1}/{self.max_retries}")
                
            except Exception as e:
                print(f"未知異常: {e}")
                return None
            
            # 等待后重試
            if retries < self.max_retries:
                time.sleep(retry_delay)
                retry_delay *= 2  # 指數(shù)退避策略
            
            retries += 1
        
        print("達(dá)到最大重試次數(shù)，放棄")
        return None
    
    def get_ip_location(self, ip_address: str) -> Optional[str]:
        """
        獲取IP地址的地理位置
        
        Args:
            ip_address: IP地址
            
        Returns:
            城市名稱或None
        """
        url = f"http://ip-api.com/json/{ip_address}"
        
        result = self.call_api_with_retry(url)
        
        if result and result.get('status') == 'success':
            return result.get('city')
        
        return None
    
    def close(self):
        """關(guān)閉會(huì)話"""
        self.session.close()
# 使用示例
if __name__ == '__main__':
    query = RobustIPQuery(timeout=10, max_retries=3)
    
    # 測(cè)試不同情況
    test_cases = [
        "8.8.8.8",  # 正常IP
        "invalid-ip",  # 無效IP
        "192.168.1.1",  # 私有IP
    ]
    
    for ip in test_cases:
        print(f"\n查詢IP: {ip}")
        city = query.get_ip_location(ip)
        print(f"結(jié)果: {city}")
    
    query.close()

import json
import os
from datetime import datetime, timedelta
from typing import Optional, Dict, Any
import threading
class IPQueryCache:
    """IP查詢結(jié)果緩存"""
    
    def __init__(self, cache_file: str = "ip_cache.json", expire_hours: int = 24):
        """
        初始化緩存
        
        Args:
            cache_file: 緩存文件路徑
            expire_hours: 緩存過期時(shí)間（小時(shí)）
        """
        self.cache_file = cache_file
        self.expire_hours = expire_hours
        self.lock = threading.Lock()
        self.cache = self._load_cache()
    
    def _load_cache(self) -> Dict[str, Any]:
        """加載緩存文件"""
        if os.path.exists(self.cache_file):
            try:
                with open(self.cache_file, 'r', encoding='utf-8') as f:
                    return json.load(f)
            except Exception:
                return {}
        return {}
    
    def _save_cache(self):
        """保存緩存到文件"""
        with self.lock:
            try:
                with open(self.cache_file, 'w', encoding='utf-8') as f:
                    json.dump(self.cache, f, ensure_ascii=False, indent=2)
            except Exception as e:
                print(f"保存緩存失敗: {e}")
    
    def get(self, ip: str) -> Optional[Dict[str, Any]]:
        """
        獲取緩存結(jié)果
        
        Args:
            ip: IP地址
            
        Returns:
            緩存的位置信息或None
        """
        with self.lock:
            if ip in self.cache:
                entry = self.cache[ip]
                
                # 檢查是否過期
                cache_time = datetime.fromisoformat(entry.get('timestamp', ''))
                if datetime.now() - cache_time < timedelta(hours=self.expire_hours):
                    return entry.get('data')
                else:
                    # 刪除過期緩存
                    del self.cache[ip]
                    self._save_cache()
        
        return None
    
    def set(self, ip: str, data: Dict[str, Any]):
        """
        設(shè)置緩存
        
        Args:
            ip: IP地址
            data: 位置信息數(shù)據(jù)
        """
        with self.lock:
            self.cache[ip] = {
                'data': data,
                'timestamp': datetime.now().isoformat()
            }
            self._save_cache()
    
    def clear_expired(self):
        """清除過期緩存"""
        with self.lock:
            expired_ips = []
            
            for ip, entry in self.cache.items():
                cache_time = datetime.fromisoformat(entry.get('timestamp', ''))
                if datetime.now() - cache_time >= timedelta(hours=self.expire_hours):
                    expired_ips.append(ip)
            
            for ip in expired_ips:
                del self.cache[ip]
            
            if expired_ips:
                self._save_cache()
    
    def get_stats(self) -> Dict[str, Any]:
        """獲取緩存統(tǒng)計(jì)信息"""
        with self.lock:
            total = len(self.cache)
            expired = 0
            
            for entry in self.cache.values():
                cache_time = datetime.fromisoformat(entry.get('timestamp', ''))
                if datetime.now() - cache_time >= timedelta(hours=self.expire_hours):
                    expired += 1
            
            return {
                'total': total,
                'active': total - expired,
                'expired': expired,
                'hit_rate': 0.0  # 需要實(shí)際統(tǒng)計(jì)
            }
# 使用緩存的IP查詢服務(wù)
class CachedIPQuery:
    """帶緩存的IP查詢服務(wù)"""
    
    def __init__(self):
        self.cache = IPQueryCache()
        self.session = requests.Session()
    
    def get_location(self, ip: str) -> Optional[Dict[str, Any]]:
        """
        獲取IP位置信息（帶緩存）
        
        Args:
            ip: IP地址
            
        Returns:
            位置信息字典
        """
        # 首先檢查緩存
        cached = self.cache.get(ip)
        if cached:
            return cached
        
        # 緩存未命中，查詢API
        url = f"http://ip-api.com/json/{ip}"
        
        try:
            response = self.session.get(url, timeout=10)
            response.raise_for_status()
            data = response.json()
            
            if data.get('status') == 'success':
                result = {
                    'ip': ip,
                    'city': data.get('city'),
                    'country': data.get('country'),
                    'isp': data.get('isp'),
                    'timestamp': datetime.now().isoformat()
                }
                
                # 存入緩存
                self.cache.set(ip, result)
                
                return result
        except Exception as e:
            print(f"查詢異常: {e}")
        
        return None
    
    def close(self):
        """關(guān)閉服務(wù)"""
        self.session.close()
        self.cache._save_cache()

class SecurityMonitor:
    """網(wǎng)絡(luò)安全監(jiān)控"""
    
    def __init__(self):
        self.ip_service = IPinfoService()
        self.suspicious_ips = set()
    
    def analyze_access_log(self, log_file: str):
        """分析訪問日志"""
        with open(log_file, 'r') as f:
            for line in f:
                # 解析日志，提取IP
                ip = self._extract_ip_from_log(line)
                
                if ip and ip not in self.suspicious_ips:
                    location = self.ip_service.get_location_info(ip)
                    
                    # 檢查可疑地區(qū)
                    if self._is_suspicious(location):
                        self.suspicious_ips.add(ip)
                        self._send_alert(ip, location)
    
    def _is_suspicious(self, location: Dict) -> bool:
        """判斷是否為可疑地區(qū)"""
        suspicious_countries = ['XX', 'YY']  # 示例國家代碼
        return location.get('country') in suspicious_countries
    
    def _send_alert(self, ip: str, location: Dict):
        """發(fā)送告警"""
        print(f"安全告警: 可疑訪問來自 {ip} ({location.get('city')})")

class UserAnalytics:
    """用戶行為分析"""
    
    def __init__(self):
        self.ip_service = IPinfoService()
        self.user_locations = {}
    
    def track_user_location(self, user_id: str, ip: str):
        """跟蹤用戶地理位置"""
        location = self.ip_service.get_location_info(ip)
        
        if user_id not in self.user_locations:
            self.user_locations[user_id] = []
        
        self.user_locations[user_id].append({
            'timestamp': datetime.now(),
            'location': location,
            'ip': ip
        })
    
    def get_user_travel_history(self, user_id: str) -> List[Dict]:
        """獲取用戶旅行歷史"""
        return self.user_locations.get(user_id, [])
    
    def analyze_user_patterns(self):
        """分析用戶行為模式"""
        # 實(shí)現(xiàn)用戶行為模式分析
        pass

class AdTargeting:
    """廣告地域定向"""
    
    def __init__(self):
        self.ip_service = IPinfoService()
    
    def get_targeted_ads(self, ip: str) -> List[Dict]:
        """獲取定向廣告"""
        location = self.ip_service.get_location_info(ip)
        
        # 根據(jù)地理位置選擇廣告
        ads = []
        
        if location.get('country') == 'CN':
            ads.append({'type': 'local', 'content': '本地服務(wù)'})
        elif location.get('country') == 'US':
            ads.append({'type': 'global', 'content': '國際產(chǎn)品'})
        
        return ads