#!/usr/bin/env python3 """ Fumamx API Client - 基于抓取的实际 API 生成 支持自动登录刷新 Token """ import json import requests import time from pathlib import Path from typing import Dict, Optional, List, Any from urllib.parse import urljoin from . import config from .auth import refresh_auth, load_tokens, AuthError class FumamxClient: """孚盟MX API 客户端 - 贸易数据模块 认证优先级: 1. 使用传入的 cookie_file 中的 cookies 2. 使用缓存的 tokens (通过 Playwright 登录获得) 3. 自动触发 Playwright 登录获取新 tokens """ BASE_URL = "https://www.fumamx.com" def __init__(self, cookie_file: str = "/home/ubuntu/fumamx_cookies.txt", auto_login: bool = True): self.cookie_file = cookie_file self.session = requests.Session() self.cookies = {} self.common_params = {} self.auto_login = auto_login self._auth_refreshed = False # 尝试加载认证信息 if not self._load_cookies() and auto_login: # Cookie 加载失败,尝试使用缓存的 token if not self._load_cached_tokens(): # 缓存也没有,触发自动登录 self._do_auto_login() self._init_common_params() def _load_cookies(self) -> bool: """从 JSON 文件加载 cookies Returns: bool: 是否成功加载 """ try: if not Path(self.cookie_file).exists(): return False with open(self.cookie_file, 'r') as f: cookie_list = json.load(f) for cookie in cookie_list: name = cookie.get('name') value = cookie.get('value') if name and value: self.cookies[name] = value self.session.cookies.set(name, value) print(f"✓ 已加载 {len(self.cookies)} 个 cookies") return True except Exception as e: print(f"⚠ 加载 cookies 失败: {e}") return False def _load_cached_tokens(self) -> bool: """从缓存加载 tokens 和 cookies Returns: bool: 是否成功加载 """ tokens = load_tokens() if not tokens: return False has_tokens = tokens.get('accessToken') and tokens.get('individualAccessToken') if has_tokens: print("✓ 使用缓存的 tokens") # 同时加载缓存的 cookies (如 _t) cached_cookies = tokens.get('cookies', {}) for name, value in cached_cookies.items(): self.cookies[name] = value self.session.cookies.set(name, value) if cached_cookies: print(f"✓ 已加载 {len(cached_cookies)} 个缓存 cookies") return has_tokens def _do_auto_login(self) -> bool: """触发自动登录获取新 tokens 和 cookies Returns: bool: 是否登录成功 """ if self._auth_refreshed: return False # 避免循环刷新 try: print("⚠ Cookie/Token 无效,正在自动登录...") result = refresh_auth() if result['success']: self._auth_refreshed = True # 加载登录获得的 cookies 到 session login_cookies = result.get('cookies', {}) for name, value in login_cookies.items(): self.cookies[name] = value self.session.cookies.set(name, value) if login_cookies: print(f"✓ 已加载 {len(login_cookies)} 个登录 cookies") # 重新加载 tokens self._load_cached_tokens() return True else: print(f"✗ 自动登录失败: {result.get('error')}") return False except AuthError as e: print(f"✗ 自动登录需要 Playwright: {e}") print(" 请运行: pip install playwright && playwright install chromium") return False except Exception as e: print(f"✗ 自动登录出错: {e}") return False def _init_common_params(self): """初始化公共参数(优先使用缓存/自动登录获取的 token)""" # 优先使用缓存的 tokens tokens = load_tokens() access_token = tokens.get('accessToken') if tokens else None individual_token = tokens.get('individualAccessToken') if tokens else None # 如果没有缓存的,使用配置文件中的 if not access_token: access_token = config.ACCESS_TOKEN if not individual_token: individual_token = config.INDIVIDUAL_ACCESS_TOKEN self.common_params = { 'accessToken': access_token, 'individualAccessToken': individual_token, '_filterCompanyId': config.COMPANY_ID, 'isV5': 'true', 'isV5AndNewMail': 'true', 'isNewArchCompany': 'true', 'languageForMX5': 'zh-cn', 'operating_terminal': 'pc_网页端', 'event_source': 'MX', } def _headers(self) -> Dict[str, str]: """构造请求头""" return { 'Accept': 'application/json, text/plain, */*', 'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8', 'Content-Type': 'application/json;charset=UTF-8', 'Origin': self.BASE_URL, 'Referer': f"{self.BASE_URL}/", 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36', 'X-Requested-With': 'XMLHttpRequest', } def _build_url(self, path: str) -> str: """构建完整 URL""" if path.startswith('http'): return path return f"{self.BASE_URL}{path}" def _merge_params(self, extra_params: Optional[Dict] = None) -> Dict: """合并公共参数和额外参数""" params = self.common_params.copy() params['_@#time'] = int(time.time() * 1000) if extra_params: params.update(extra_params) # 过滤 None 值 return {k: v for k, v in params.items() if v is not None} def get(self, path: str, params: Optional[Dict] = None) -> Dict: """GET 请求""" url = self._build_url(path) merged_params = self._merge_params(params) response = self.session.get( url, headers=self._headers(), params=merged_params, timeout=30 ) response.raise_for_status() return response.json() def post(self, path: str, data: Optional[Dict] = None) -> Dict: """POST 请求""" url = self._build_url(path) merged_params = self._merge_params() response = self.session.post( url, headers=self._headers(), params=merged_params, json=data, timeout=30 ) response.raise_for_status() return response.json() # ============ 贸易数据 - 公司搜索 ============ def search_company(self, keyword: str, page: int = 1, page_size: int = 10) -> Dict: """搜索公司/企业 对应端点: POST /pcapi/findServer/standard/company """ data = { "selectedKey": [], "searchType": "all", "activeType": "company", "inputValue": keyword, "keyword": [keyword], "FyStr": keyword, "productFy": [keyword], "currentValue": "input", "language": ["en", "ru", "pt", "spa"], "selectCountry": [], "searchMode": 0, "country": [], "industry": [], "employeesLevel": [], "founded": [], "priceRange": [], "filterCust": 0, "filterView": 0, "excludeGovernment": False, "excludeSchool": False, "excludeForwarder": False, "excludeBank": False, "excludeCn": True, "page": page, "size": page_size } return self.post("/pcapi/findServer/standard/company", data) # ============ 贸易数据 - 进口企业 ============ def import_enterprise(self, company_name: str, country: str = "") -> Dict: """获取进口企业数据 对应端点: POST /pcapi/tradeData/api/V2/TradeCompany/ImpEnterprise """ data = { "companyName": company_name, "country": country, "page": 1, "pageSize": 20 } return self.post("/pcapi/tradeData/api/V2/TradeCompany/ImpEnterprise", data) def import_enterprise_stats(self, company_name: str) -> Dict: """获取进口企业统计 对应端点: POST /pcapi/tradeData/api/V1/TradeCompany/ImpEnterpriseStatistics """ data = {"companyName": company_name} return self.post("/pcapi/tradeData/api/V1/TradeCompany/ImpEnterpriseStatistics", data) # ============ 贸易数据 - 出口企业 ============ def export_enterprise(self, company_name: str, country: str = "") -> Dict: """获取出口企业数据 对应端点: POST /pcapi/tradeData/api/V2/TradeCompany/ExpEnterprise """ data = { "companyName": company_name, "country": country, "page": 1, "pageSize": 20 } return self.post("/pcapi/tradeData/api/V2/TradeCompany/ExpEnterprise", data) def export_enterprise_stats(self, company_name: str) -> Dict: """获取出口企业统计 对应端点: POST /pcapi/tradeData/api/V1/TradeCompany/ExpEnterpriseStatistics """ data = {"companyName": company_name} return self.post("/pcapi/tradeData/api/V1/TradeCompany/ExpEnterpriseStatistics", data) # ============ 贸易数据 - 产品分析 ============ def company_agg_data(self, company_name: str) -> Dict: """获取公司聚合数据 对应端点: POST /pcapi/tradeData/api/V2/TradeProduct/getCompanyAggData """ data = {"companyName": company_name} return self.post("/pcapi/tradeData/api/V2/TradeProduct/getCompanyAggData", data) def hs_code_analysis(self, hs_code: str) -> Dict: """HS 编码分析 对应端点: POST /pcapi/tradeData/api/V2/TradeProduct/HsCodeAnalysis """ data = {"hsCode": hs_code} return self.post("/pcapi/tradeData/api/V2/TradeProduct/HsCodeAnalysis", data) def export_enterprise_analysis(self, company_name: str) -> Dict: """出口企业分析 对应端点: POST /pcapi/tradeData/api/V2/TradeProduct/ExpEnterpriseAnalysis """ data = {"companyName": company_name} return self.post("/pcapi/tradeData/api/V2/TradeProduct/ExpEnterpriseAnalysis", data) def company_analysis_agg(self, company_name: str) -> Dict: """公司分析聚合数据 对应端点: POST /pcapi/tradeData/api/V2/TradeProduct/getCompanyAnalysisAggData """ data = {"companyName": company_name} return self.post("/pcapi/tradeData/api/V2/TradeProduct/getCompanyAnalysisAggData", data) def export_compare_lose_analysis(self, company_name: str, compare_company: str) -> Dict: """出口对比流失分析 对应端点: POST /pcapi/tradeData//api/V2/TradeProduct/ExpEnterpriseCompareLoseAnalysis """ data = { "companyName": company_name, "compareCompany": compare_company } return self.post("/pcapi/tradeData//api/V2/TradeProduct/ExpEnterpriseCompareLoseAnalysis", data) def export_compare_new_analysis(self, company_name: str, compare_company: str) -> Dict: """出口对比新增分析 对应端点: POST /pcapi/tradeData//api/V2/TradeProduct/ExpEnterpriseCompareNewAnalysis """ data = { "companyName": company_name, "compareCompany": compare_company } return self.post("/pcapi/tradeData//api/V2/TradeProduct/ExpEnterpriseCompareNewAnalysis", data) def export_country_analysis(self, country: str) -> Dict: """出口国家分析 对应端点: POST /pcapi/tradeData/api/V2/TradeProduct/ExpCountryAnalysis """ data = {"country": country} return self.post("/pcapi/tradeData/api/V2/TradeProduct/ExpCountryAnalysis", data) def country_analysis_data(self, country: str) -> Dict: """国家分析数据 对应端点: POST /pcapi/tradeData/api/V2/TradeProduct/getCountryAnalysisData """ data = {"country": country} return self.post("/pcapi/tradeData/api/V2/TradeProduct/getCountryAnalysisData", data) # ============ 贸易详情 ============ def trade_detail_full(self, trade_no: str, source_country: str = "", reg_date: str = "", in_out: str = "") -> Dict: """获取完整贸易详情 对应端点: GET /pcapi/tradeData/api/V2/Trade/TradeDetailFull """ return self.get("/pcapi/tradeData/api/V2/Trade/TradeDetailFull", { "tradeNo": trade_no, "sourceCountry": source_country, "regDate": reg_date, "inOut": in_out, "sourceType": "1" }) # ============ 用户/账户 ============ def get_trade_source_countries(self) -> Dict: """获取贸易来源国家列表 对应端点: GET /pcapi/tradeData/api/V2/Account/GetTradeSourceCountrys """ return self.get("/pcapi/tradeData/api/V2/Account/GetTradeSourceCountrys") # ============ 企业详情 API ============ def get_company_info(self, company_id: str, domain: str) -> Dict: """获取公司详细信息 对应端点: GET /pcapi/findServer/find/getMxCompanyInfo 包含: 地址、电话、员工数、行业等信息 """ params = { "id": company_id, "domain": domain, } return self.get("/pcapi/findServer/find/getMxCompanyInfo", params) def get_company_contacts(self, company_id: str, domain: str) -> Dict: """获取公司联系人列表 对应端点: GET /pcapi/findServer/find/mxStaffList 包含: 姓名、邮箱、职位、LinkedIn、电话等 """ params = { "id": company_id, "domain": domain, } return self.get("/pcapi/findServer/find/mxStaffList", params) def get_company_phones(self, domain: str) -> Dict: """获取公司电话/图片信息 对应端点: GET /pcapi/findServer/find/spiderDomainImage 包含: 电话、图片等爬虫抓取的信息 """ params = { "domain": domain, } return self.get("/pcapi/findServer/find/spiderDomainImage", params) # ============ 其他 ============ def wap_label_list(self, wid: str = "330204727296") -> Dict: """获取标签列表 对应端点: GET /pcapi/v1/waplabellist """ return self.get("/pcapi/v1/waplabellist", { "wid": wid, "format": "mx" }) def check_sse_exist(self) -> Dict: """检查 SSE 是否存在 对应端点: GET /v1/isSseExist """ return self.get("/v1/isSseExist") # ============ 通用 API 探测 ============ def probe_api(self, endpoint: str) -> Dict: """探测任意 API 端点""" try: return self.get(endpoint) except Exception as e: return {"error": str(e), "endpoint": endpoint} def call_api(self, method: str, path: str, data: Optional[Dict] = None, params: Optional[Dict] = None) -> Dict: """通用 API 调用""" if method.upper() == "GET": return self.get(path, params) elif method.upper() == "POST": return self.post(path, data) else: raise ValueError(f"不支持的 HTTP 方法: {method}") def check_auth(self) -> bool: """检查当前认证是否有效 Returns: bool: 认证是否有效 """ try: result = self.search_company("test", page=1, page_size=1) return result.get("code") == "0" except Exception: return False def refresh_if_needed(self) -> bool: """如果需要,刷新认证 Returns: bool: 刷新后认证是否有效 """ if self.check_auth(): return True if not self.auto_login: return False print("⚠ 认证失效,尝试自动刷新...") if self._do_auto_login(): self._init_common_params() # 重新初始化参数 return self.check_auth() return False