Преглед на файлове

v0.8(海外系统接口修改为api key接入)

kyle преди 1 седмица
родител
ревизия
387d65d546

+ 1 - 3
auto-generate-export-contracts/.env

@@ -1,5 +1,3 @@
 # Overseas Price API Configuration
-# Copy this file and fill in your credentials
 PRICE_API_URL=https://overseasprice.weleads.cn
-PRICE_API_USERNAME=haiwai
-PRICE_API_PASSWORD=112233
+PRICE_API_KEY=ow_GF-HRlFrAdJnulmO5cgqglE9Av5BSNHgTQnl1pA_gqc

+ 1 - 1
auto-generate-export-contracts/SKILL.md

@@ -183,7 +183,7 @@ result = generate_contracts(order_text, output_dir='.', quotation_path='报价
 - `assets/vehicle-sales-contract-template.docx`: 销售合同模板
 - `scripts/overseas_price_client.py`: 海外价格系统 API 客户端
 - `scripts/parse_quotation.py`: 报价表 xlsx 解析引擎
-- `.env`: API 连接配置(URL、用户名、密码
+- `.env`: API 连接配置(URL、API Key
 
 ## 字段映射详情
 

BIN
auto-generate-export-contracts/scripts/__pycache__/match_vehicle.cpython-313.pyc


BIN
auto-generate-export-contracts/scripts/__pycache__/overseas_price_client.cpython-313.pyc


+ 29 - 67
auto-generate-export-contracts/scripts/overseas_price_client.py

@@ -4,17 +4,14 @@
 
 Replaces the Excel-based price config by directly querying
 the overseas price system API.
-Mirrors the Java OverseasPriceApi behavior.
 
 Configuration: env vars or .env file.
-    PRICE_API_URL      - API base URL
-    PRICE_API_USERNAME - login username
-    PRICE_API_PASSWORD - login password
+    PRICE_API_URL  - API base URL
+    PRICE_API_KEY  - API key for authentication
 """
 
 import os
 import logging
-from datetime import datetime
 from typing import Any, Optional
 
 import requests
@@ -40,8 +37,7 @@ def _load_config():
                     os.environ[key] = val
     return {
         'url': os.environ.get('PRICE_API_URL', ''),
-        'username': os.environ.get('PRICE_API_USERNAME', ''),
-        'password': os.environ.get('PRICE_API_PASSWORD', ''),
+        'api_key': os.environ.get('PRICE_API_KEY', ''),
     }
 
 
@@ -50,82 +46,28 @@ def get_client():
     if not cfg['url']:
         raise ValueError(
             'Overseas price API not configured. '
-            'Set PRICE_API_URL / PRICE_API_USERNAME / PRICE_API_PASSWORD in .env'
+            'Set PRICE_API_URL / PRICE_API_KEY in .env'
         )
-    return OverseasPriceClient(cfg['url'], cfg['username'], cfg['password'])
+    return OverseasPriceClient(cfg['url'], cfg['api_key'])
 
 
 class OverseasPriceClient:
     """API client for the overseas price system."""
 
-    _COOKIE_EXPIRE_SECONDS = 24 * 60 * 60
-
-    def __init__(self, base_url, username, password):
+    def __init__(self, base_url, api_key):
         self._base_url = base_url.rstrip('/')
-        self._username = username
-        self._password = password
+        self._api_key = api_key
         self._session = requests.Session()
-        self._cookie_header = None
-        self._cookie_expire_at = 0
-
-    def _is_session_alive(self):
-        now_ts = datetime.now().timestamp()
-        return bool(self._cookie_header) and now_ts < self._cookie_expire_at
-
-    def ensure_login(self):
-        if self._is_session_alive():
-            return
-        self._login()
-
-    def _login(self):
-        url = '{}/login'.format(self._base_url)
         # Suppress SSL warnings for self-signed certs
         import urllib3
         urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
-        body = {'username': self._username, 'password': self._password, 'remember': True}
-        resp = self._session.post(url, json=body, timeout=30, verify=False)
-        resp.raise_for_status()
-        login_data = resp.json()
-        code = login_data.get('code', -1)
-        if code != 0:
-            raise RuntimeError('Login failed (code={}): {}'.format(code, login_data.get('message', '')))
-        cookie_parts = []
-        for cookie in resp.cookies:
-            cookie_parts.append('{}={}'.format(cookie.name, cookie.value))
-        self._cookie_header = '; '.join(cookie_parts)
-        self._cookie_expire_at = datetime.now().timestamp() + self._COOKIE_EXPIRE_SECONDS
-        if not self._cookie_header:
-            raise RuntimeError('Login failed: no cookie returned')
-
-    def search_cars(self, keyword, page=1, per_page=20):
-        self.ensure_login()
-        url = '{}/api/cars'.format(self._base_url)
-        params = {'keyword': keyword, 'page': page, 'per_page': per_page}
-        data = self._request('GET', url, params=params)
-        cars = self._extract_cars_node(data)
-        return cars if isinstance(cars, list) else []
-
-    def get_countries(self):
-        self.ensure_login()
-        url = '{}/api/countries'.format(self._base_url)
-        data = self._request('GET', url)
-        countries = self._extract_countries_node(data)
-        return countries if isinstance(countries, list) else []
-
-    def get_cars_by_country(self, country_id, vehicle_type=None):
-        self.ensure_login()
-        url = '{}/api/countries/{}/cars'.format(self._base_url, country_id)
-        params = {}
-        if vehicle_type:
-            params['keyword'] = vehicle_type
-        return self._request('GET', url, params=params)
 
     def _request(self, method, url, **kwargs):
         headers = kwargs.pop('headers', {})
         headers.setdefault('Content-Type', 'application/json')
         headers.setdefault('Accept', 'application/json')
-        if self._cookie_header:
-            headers['Cookie'] = self._cookie_header
+        if self._api_key:
+            headers['Authorization'] = 'Bearer {}'.format(self._api_key)
         kwargs['headers'] = headers
         kwargs.setdefault('timeout', 60)
         kwargs.setdefault('verify', False)
@@ -149,6 +91,26 @@ class OverseasPriceClient:
             return result.get('data')
         return result
 
+    def search_cars(self, keyword, page=1, per_page=20):
+        url = '{}/api/cars'.format(self._base_url)
+        params = {'keyword': keyword, 'page': page, 'per_page': per_page}
+        data = self._request('GET', url, params=params)
+        cars = self._extract_cars_node(data)
+        return cars if isinstance(cars, list) else []
+
+    def get_countries(self):
+        url = '{}/api/countries'.format(self._base_url)
+        data = self._request('GET', url)
+        countries = self._extract_countries_node(data)
+        return countries if isinstance(countries, list) else []
+
+    def get_cars_by_country(self, country_id, vehicle_type=None):
+        url = '{}/api/countries/{}/cars'.format(self._base_url, country_id)
+        params = {}
+        if vehicle_type:
+            params['keyword'] = vehicle_type
+        return self._request('GET', url, params=params)
+
     @staticmethod
     def _extract_list_from_data(data):
         if data is None: