87 lines
3.0 KiB
Python
87 lines
3.0 KiB
Python
|
|
import requests
|
|
import json
|
|
import time
|
|
import logging
|
|
import sys
|
|
from requests.adapters import HTTPAdapter
|
|
from urllib3.util.retry import Retry
|
|
|
|
# Configure logger
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class IFindClient:
|
|
"""
|
|
iFinD HTTP API Client Base Class.
|
|
Handles authentication and token management.
|
|
"""
|
|
def __init__(self, refresh_token: str):
|
|
self.refresh_token = refresh_token
|
|
self.access_token = None
|
|
self.token_expiry = 0
|
|
self.base_url = "https://quantapi.51ifind.com/api/v1"
|
|
|
|
# Initialize Session with Retry logic
|
|
self.session = requests.Session()
|
|
retries = Retry(
|
|
total=5,
|
|
backoff_factor=1,
|
|
status_forcelist=[500, 502, 503, 504],
|
|
allowed_methods=["POST", "GET"]
|
|
)
|
|
self.session.mount('https://', HTTPAdapter(max_retries=retries))
|
|
|
|
def _get_access_token(self):
|
|
"""Get or refresh access_token"""
|
|
if self.access_token and time.time() < self.token_expiry:
|
|
return self.access_token
|
|
|
|
url = f"{self.base_url}/get_access_token"
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"refresh_token": self.refresh_token
|
|
}
|
|
|
|
try:
|
|
response = self.session.post(url, headers=headers)
|
|
data = response.json()
|
|
if data.get("errorcode") == 0:
|
|
self.access_token = data["data"]["access_token"]
|
|
# Expires in 7 days, refresh every 6 days
|
|
self.token_expiry = time.time() + (6 * 24 * 3600)
|
|
return self.access_token
|
|
else:
|
|
error_msg = f"iFinD access token error: {data.get('errmsg')} (Code: {data.get('errorcode')})"
|
|
if data.get("errorcode") == -1301:
|
|
logger.critical(f"CRITICAL ERROR: {error_msg}")
|
|
logger.critical("Please check your IFIND_REFRESH_TOKEN in .env file.")
|
|
# In backend service, we shouldn't sys.exit() as it kills the server
|
|
# Just raise exception
|
|
raise Exception(error_msg)
|
|
raise Exception(error_msg)
|
|
except Exception as e:
|
|
logger.error(f"Error refreshing iFinD access token: {e}")
|
|
return None
|
|
|
|
def post(self, endpoint: str, params: dict):
|
|
"""Send POST request to API endpoint"""
|
|
token = self._get_access_token()
|
|
if not token:
|
|
return None
|
|
|
|
url = f"{self.base_url}/{endpoint}"
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"access_token": token
|
|
}
|
|
|
|
try:
|
|
response = self.session.post(url, headers=headers, json=params, timeout=30)
|
|
if response.status_code != 200:
|
|
logger.error(f"iFinD API HTTP Error: {response.status_code} - {response.text}")
|
|
return None
|
|
return response.json()
|
|
except Exception as e:
|
|
logger.error(f"iFinD API Request Error on {endpoint}: {e}")
|
|
return None
|