jiangcang_vj/descending_cabin_task.py

"""
询价 + 验价任务：先询价，按航班号/舱位/行李匹配 result 取 data，再验价并按要求处理时效与重试。
"""
import os
import json
import time
import traceback
import requests
import threading
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor, as_completed
from xmi_logger import XmiLogger
import csv


# 询价\验价
BASE_URL = "http://8.218.51.130:9000"
CID = "cba37f642eab11ef8e7200163e01a06e"
# 汇率
RATES_URL = "http://8.218.51.130:9003/api/v1/rates"
RATES_CID = "750B5141EDBF7FA6F73A99C768130099"
# 政策
# POLICY_URL = "http://192.168.20.134:8787/prediction/rules/sync"
POLICY_URL = "http://direct.ysjipiao.com:8787/prediction/rules/sync"

VALID_DURATION_SECONDS = 3 * 60  # 3 分钟 验价结果有效时间 
RETRY_TIMES = 3  # 请求询价/验价接口重试次数
RETRY_INTERVAL = 10  # 秒 询价/验价接口重试间隔
RETRY_DURATION = 3 * 60  # 持续 3 分钟 询价/验价接口重试时间
PRICE_DIFF_THRESHOLD = 10  # 验价与任务期望价格允许的差值（已统一币种后），在此范围内才视为通过  人民币
RATES_RETRY_TIMES = 3  # 汇率接口重试次数
RATES_RETRY_INTERVAL = 2  # 汇率接口重试间隔



def time_format_conversion(match_time,
                           in_strftime_str="%Y-%m-%dT%H:%M:%S",
                           out_strftime_str="%Y%m%d%H%M%S"):
    """
    时间格式转换
    match_time 输入的时间 023-07-29T04:15:00
    in_strftime_str 输入的时间格式
    out_strftime_str 输出的时间格式
    """
    time_array = time.strptime(match_time, in_strftime_str)
    return time.strftime(out_strftime_str, time_array)




class FlightPriceRequestError(Exception):
    """请求询价/验价接口时的网络或响应异常."""
    def __init__(self, message: str, cause: Exception | None = None):
        self.cause = cause
        super().__init__(message)



def fetch_rate(base: str, symbols: str, rates_url: str = RATES_URL, cid: str = RATES_CID) -> float | None:
    """
    请求汇率接口，获取 1 单位 base 兑换为 symbols 的汇率。
    响应格式: {"code": 0, "msg": "success", "data": {"base": "USD", "symbols": "CNY", "rate": 6.8437}}。
    带重试，失败返回 None。
    """
    last_err: Exception | None = None
    for attempt in range(RATES_RETRY_TIMES):
        try:
            resp = requests.get(
                rates_url,
                params={"base": base, "symbols": symbols},
                headers={
                    "cid": cid,
                    "User-Agent": "Apifox/1.0.0 (https://apifox.com)",
                },
                timeout=15 
            )
            resp.raise_for_status()
            body = resp.json()
            if body.get("code") != 0:
                last_err = ValueError(body.get("msg", "汇率接口返回非成功"))
                if attempt < RATES_RETRY_TIMES - 1:
                    time.sleep(RATES_RETRY_INTERVAL)
                continue
            data = body.get("data")
            if isinstance(data, dict) and "rate" in data:
                return float(data["rate"])
            last_err = ValueError("响应缺少 data.rate")
        except (requests.RequestException, json.JSONDecodeError, TypeError, ValueError, KeyError) as e:
            last_err = e
        if attempt < RATES_RETRY_TIMES - 1:
            time.sleep(RATES_RETRY_INTERVAL)
    return None


class FlightPriceClient:
    """请求封装：询价、验价接口."""

    def __init__(self, base_url: str = BASE_URL, cid: str = CID):
        
        self.base_url = base_url.rstrip("/")
        self.cid = cid
        self.headers = {
            "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36",
            "Content-Type": "application/json",
        }

    def _request(self, url: str, payload: dict) -> dict:
        """发起 POST 请求并解析 JSON，统一处理异常。失败时按 RETRY_TIMES 次重试，间隔 RETRY_INTERVAL 秒."""
        last_err = None
        # print(json.dumps(payload, ensure_ascii=False, indent=2))
        for attempt in range(RETRY_TIMES):
            try:
                if attempt > 0:
                    time.sleep(RETRY_INTERVAL)
                resp = requests.post(url, headers=self.headers, json=payload, timeout=30)
                resp.raise_for_status()
                body = resp.json()
                print(json.dumps(body, ensure_ascii=False)[:200])
                return body
                # print(resp.json())
                # return resp.json()
            except requests.Timeout as e:
                last_err = FlightPriceRequestError(f"请求超时: {url}", cause=e)
            except requests.ConnectionError as e:
                last_err = FlightPriceRequestError(f"连接失败: {url}", cause=e)
            except requests.HTTPError as e:
                last_err = FlightPriceRequestError(
                    f"HTTP 错误 {getattr(e.response, 'status_code', '?')}: {url}", cause=e
                )
            except requests.RequestException as e:
                last_err = FlightPriceRequestError(f"请求异常: {url} - {e}", cause=e)
            except json.JSONDecodeError as e:
                last_err = FlightPriceRequestError(f"响应非合法 JSON: {url}", cause=e)
        raise last_err

    def search_flights(
        self,
        from_city_code: str,
        to_city_code: str,
        from_day: str,
    ) -> dict:
        """询价接口."""
        url = f"{self.base_url}/v1/search_flights"
        payload = {
            "from_city_code": from_city_code,
            "to_city_code": to_city_code,
            "from_day": from_day,
            "cid": self.cid,
        }
        return self._request(url, payload)

    def verify_price(
        self,
        from_city_code: str,
        to_city_code: str,
        from_day: str,
        data: str,
        not_verify: bool = False,
        async_: bool = True,
    ) -> dict:
        """验价接口."""
        url = f"{self.base_url}/v1/verify_price"
        payload = {
            "from_city_code": from_city_code,
            "to_city_code": to_city_code,
            "from_day": from_day,
            "data": data,
            "cid": self.cid,
            "not_verify": not_verify,   
            "async": async_,
        }
        return self._request(url, payload)


class ResultMatcher:
    """从询价 result 中按航班号、舱位、行李匹配一条，并取出 data."""

    @staticmethod
    def match(
        result: list[dict],
        cabins: str,
        baggages: str,
        flight_numbers: str,
    ) -> dict | None:
        """
        在 result 中匹配：按段顺序一致。cabins / baggages / flight_numbers 均为 ";" 分隔的多段字符串，
        第 i 段需满足：seg[i].cabin == 第 i 个舱位、seg[i].baggage == 第 i 个行李、seg[i].flight_number == 第 i 个航班号。
        返回匹配到的那条 result 项（含 data），未匹配到返回 None。
        """
        separator = '|'  # 更换分隔符由;为|
        cabin_list = [s.strip() for s in (cabins or "").split(separator)]
        baggage_list = [s.strip() for s in (baggages or "").split(separator)]
        flight_list = [s.strip() for s in (flight_numbers or "").split(separator)] if flight_numbers else []

        n = len(cabin_list)
        if n == 0 or len(baggage_list) != n:
            return None
        if flight_list and len(flight_list) != n:
            return None

        for item in result:
            segments = item.get("segments") or []
            if len(segments) != n:
                continue
            for i in range(n):
                seg = segments[i]
                if seg.get("cabin") != cabin_list[i] or seg.get("baggage") != baggage_list[i]:
                    break
                if flight_list and seg.get("flight_number") != flight_list[i]:
                    break
            else:
                return item
        return None


class VerifyResultHandler:
    """验价返回 result 只有一条时：判断 now_time / verify_time 是否在 3 分钟内，并取价格."""

    @staticmethod
    def parse_time(ts: str | int | None) -> datetime | None:
        """解析时间字符串或时间戳."""
        if ts is None:
            return None
        if isinstance(ts, (int, float)):
            return datetime.fromtimestamp(ts)
        for fmt in ("%Y-%m-%d %H:%M:%S", "%Y%m%d%H%M%S", "%Y-%m-%dT%H:%M:%S"):
            try:
                return datetime.strptime(str(ts).strip(), fmt)
            except (ValueError, TypeError):
                continue
        return None

    @classmethod
    def is_within_3_minutes(cls, ts: str | int | None) -> bool:
        """判断给定时间是否在距今 3 分钟内."""
        dt = cls.parse_time(ts)
        if dt is None:
            return False
        return (datetime.now() - dt).total_seconds() <= VALID_DURATION_SECONDS

    @classmethod
    def get_valid_price(cls, single_result: dict) -> dict | None:
        """
        单条 result：若存在 now_time 且在 3 分钟内则返回该条（含价格）；
        否则看 verify_time 是否在 3 分钟内。
        符合条件返回该条 dict，否则返回 None。
        """
        if not single_result:
            return None
        if "now_time" in single_result and cls.is_within_3_minutes(single_result["now_time"]):
            return single_result
        if "verify_time" in single_result and cls.is_within_3_minutes(single_result["verify_time"]):
            return single_result
        return None


class FlightPriceTaskRunner:
    """单任务流程：询价 -> 匹配 -> 验价 -> 按规则取价或 not_verify 重试."""

    def __init__(self, client: FlightPriceClient | None = None, logger: XmiLogger | None = None):
        self.logger = logger or XmiLogger("flight_price_task")
        self.client = client or FlightPriceClient()
        self.matcher = ResultMatcher()
        self.handler = VerifyResultHandler()
        self.rate = fetch_rate("USD", "CNY")
        

    def run(
        self,
        task: dict,
        do_verify: bool = True,
    ) -> dict:
        """
        执行单条任务。task 需含: from_city_code, to_city_code, from_day, cabin, baggage, adult_total_price。
        flight_number 可选，用于匹配。
        do_verify=False 时仅执行询价+匹配并返回 matched（不做验价）
        返回: {"status": "ok"|"placeholder"|"no_match", "price_info": {...}, "raw_verify": ..., "raw_search": ..., "matched": ...}
        """
        from_city_code = task["from_city_code"]
        to_city_code = task["to_city_code"]
        from_day = task["from_day"]
        flight_numbers = task.get("flight_numbers")
        cabins = task["cabins"]
        baggages = task["baggages"]

        # 1. 询价
        try:
            search_resp = self.client.search_flights(from_city_code, to_city_code, from_day)
        except FlightPriceRequestError as e:
            self.logger.warning(f"询价请求异常: {e}")
            return {"status": "request_error", "msg": str(e), "phase": "search"}

        if search_resp.get("code") != 0:
            return {"status": "search_failed", "msg": search_resp.get("msg"), "raw_search": search_resp}

        result_list = search_resp.get("result") or []
        # print(result_list)
        matched = self.matcher.match(result_list, cabins=cabins, baggages=baggages, flight_numbers=flight_numbers)
        # print(matched)
        if not matched:
            return {"status": "no_match", "msg": "未匹配到航班/舱位/行李", "raw_search": search_resp}

        data = matched.get("data")
        if not data:
            return {"status": "no_data", "msg": "匹配项无 data", "raw_search": search_resp}

        # 只询价不验价走的流程
        if not do_verify:
            # return {"status": "ok", "raw_search": search_resp, "matched": matched, "data": data}
            expected_in_currency, rate_err = self._expected_price_in_verify_currency(task, matched)
            if rate_err:
                return {
                    "status": "rate_error",
                    "msg": rate_err,
                    "raw_search": search_resp,
                    "matched": matched,
                    "data": data,
                }
            actual = matched.get("adult_total_price")   # 询价和验价接口出来的币种已经是人民币, 不用再转换
            if self._price_within_threshold(expected_in_currency, actual):  # 对比
                return {
                    "status": "ok",
                    "price_info": self._extract_price_info(matched),
                    "raw_search": search_resp,
                    "matched": matched,
                    "data": data,
                }
            return {
                "status": "price_not_within_threshold",
                "msg": "询价结果价格不在阈值内",
                "expected": expected_in_currency,
                "actual": actual,
                "raw_search": search_resp,
                "matched": matched,
                "data": data,
            }
            
        # 2. 验价（先 not_verify=False）
        try:
            verify_resp = self.client.verify_price(
                from_city_code=from_city_code,
                to_city_code=to_city_code,
                from_day=from_day,
                data=data,
                not_verify=False,
                async_=True,
            )
        except FlightPriceRequestError as e:
            self.logger.warning(f"验价请求异常: {e}")
            return {"status": "request_error", "msg": str(e), "phase": "verify"}

        if verify_resp.get("code") != 0:
            return {"status": "verify_failed", "msg": verify_resp.get("msg"), "raw_verify": verify_resp}
        # print(verify_resp)
        verify_result = verify_resp.get("result")
        if isinstance(verify_result, list) and len(verify_result) >= 1:
            single = verify_result[0]
            valid = self.handler.get_valid_price(single)
            if valid is not None:
                expected_in_verify_currency, rate_err = self._expected_price_in_verify_currency(task, valid)
                if rate_err:
                    return {"status": "rate_error", "msg": rate_err}
                if self._price_within_threshold(expected_in_verify_currency, valid.get("adult_total_price")):
                    return {
                        "status": "ok",
                        "price_info": self._extract_price_info(valid),
                        "raw_verify": verify_resp,
                    }

        # 3. 不符合时效：用 not_verify=True 每 10 秒重试，最多 3 分钟
        deadline = time.monotonic() + RETRY_DURATION
        while time.monotonic() < deadline:
            time.sleep(RETRY_INTERVAL)
            try:
                verify_resp = self.client.verify_price(
                    from_city_code=from_city_code,
                    to_city_code=to_city_code,
                    from_day=from_day,
                    data=data,
                    not_verify=True,
                    async_=True,
                )
            except FlightPriceRequestError as e:
                self.logger.warning(f"验价重试请求异常: {e}")
                continue
            if verify_resp.get("code") != 0:
                continue
            verify_result = verify_resp.get("result")
            if isinstance(verify_result, list) and len(verify_result) >= 1:
                single = verify_result[0]
                valid = self.handler.get_valid_price(single)
                if valid is not None:
                    expected_in_verify_currency, rate_err = self._expected_price_in_verify_currency(task, valid)
                    if rate_err:
                        continue
                    if self._price_within_threshold(expected_in_verify_currency, valid.get("adult_total_price")):
                        return {
                            "status": "ok",
                            "price_info": self._extract_price_info(valid),
                            "raw_verify": verify_resp,
                        }
                    else:
                        # 价格不符合，直接跳出循环，不用继续校验
                        break

        # 3 分钟内都没有符合规则的数据，先占位
        return {
            "status": "placeholder",
            "msg": f"{RETRY_DURATION / 60} 分钟内未得到符合规则的价格，等待后续逻辑处理",
            "task": task,
            "data": data,
            "raw_verify": verify_resp,
        }

    def _expected_price_in_verify_currency(
        self, task: dict, valid: dict
    ) -> tuple[float | None, str | None]:
        """
        将任务期望价格换算为验价结果币种后的数值。
        返回 (换算后的价格, None)，若币种一致则直接换算为 float；
        若需汇率且获取失败返回 (None, "汇率获取失败")。
        """
        try:
            expected_val = float(task.get("adult_total_price"))
        except (TypeError, ValueError):
            return None, "任务 adult_total_price 无效"
        if self.rate is None:
            task_currency = (task.get("currency") or "USD").strip().upper()
            verify_currency = (valid.get("currency") or "CNY").strip().upper()
            if task_currency == verify_currency:
                return expected_val, None
            rate = fetch_rate(task_currency, verify_currency)
            if rate is None:
                return None, "汇率获取失败"
            self.rate = rate
        return expected_val * self.rate, None

    @staticmethod
    def _price_within_threshold(
        expected: str | int | float | None,
        actual: str | int | float | None,
        threshold: float | None = None,
    ) -> bool:
        """校验验价结果价格与任务期望价格差是否在阈值内（两者已统一币种）。阈值默认 PRICE_DIFF_THRESHOLD。"""
        if threshold is None:
            threshold = PRICE_DIFF_THRESHOLD
        try:
            e = float(expected) if expected is not None else None
            a = float(actual) if actual is not None else None
        except (TypeError, ValueError):
            return False
        if e is None or a is None:
            return False
        return abs(a - e) <= threshold

    @staticmethod
    def _extract_price_info(item: dict) -> dict:
        """从单条 result 项提取价格相关信息."""
        return {
            "adult_price": item.get("adult_price"),
            "adult_tax": item.get("adult_tax"),
            "adult_total_price": item.get("adult_total_price"),
            "currency": item.get("currency"),
            "now_time": item.get("now_time"),
            "verify_time": item.get("verify_time"),
        }


def _process_one_task(row, runner):
    """处理单条任务：构建 end_task、执行 run、解析结果。成功返回 flight_data 字典，失败返回 None。"""
    task = row
    separator = '|'  # 分隔符由;更换为|

    thread_name = threading.current_thread().name
    # print(f"[thread_name: {thread_name}] 正在处理任务: {task}")

    from_city_code, to_city_code = task["city_pair"].split("-")
    from_day = task["flight_day"].replace("-", "")

    flight_numbers = task["flight_number_1"].strip()
    if task["flight_number_2"].strip() != "VJ":
        flight_numbers += separator + task["flight_number_2"].strip()
    cabins = separator.join(["Y"] * len(flight_numbers.split(separator)))

    if str(task['baggage']) == '0':
        baggage_str = "-;-;-;-"
    else:
        baggage_str = f"1-{task['baggage']}"

    baggages = separator.join([baggage_str] * len(flight_numbers.split(separator)))

    end_task = {
        "from_city_code": from_city_code,
        "to_city_code": to_city_code,
        "from_day": from_day,
        "flight_numbers": flight_numbers,
        "cabins": cabins,
        "baggages": baggages,
        "adult_total_price": task.get("adult_total_price"),
        "currency": task.get("currency", "USD"),
    }
    # print("--------------------------------")
    # print(end_task)
    # print("--------------------------------")

    time.sleep(1)
    out = runner.run(end_task, do_verify=False)  # 不验价，仅询价
    # print(json.dumps(out, ensure_ascii=False, indent=2))
    if out.get("status") != "ok":
        # print(f"[thread_name={thread_name}] 错误: {out.get('msg')}")
        return None

    # print(f"价格: {out.get('price_info').get('adult_total_price')}")
    raw_verify = out.get("raw_verify")
    if raw_verify:
        results = raw_verify.get("result") or []
    else:
        matched = out.get("matched") or {}
        results = [matched] if matched else []
    if not results:
        return None

    print("raw_verify pass")

    # task 存放了 keep_info 的全部字段
    drop_price_change_upper = float(task.get("drop_price_change_upper"))   # 降价的最小幅度
    drop_price_change_lower = float(task.get("drop_price_change_lower"))

    max_threshold = round(drop_price_change_upper * runner.rate * 0.8)   # 降价阈值要按汇率转人民币（四舍五入到整数）

    result = results[0]
    # adult_price = result.get("adult_price")
    # adult_tax = result.get("adult_tax")
    # adult_total_price = result.get("adult_total_price")
    segments = result.get("segments") or []
    if not segments:
        return None
    end_segments = []
    baggage = segments[0].get("baggage")
    if baggage == "-;-;-;-":
        pc, kg = 0, 0   # 无行李的设置？
    else:
        pc, kg = [int(i) for i in baggage.split("-")]
    for seg in segments:
        flight_number = seg.get("flight_number")
        operating_flight_number = seg.get("operating_flight_number")
        if flight_number == operating_flight_number:
            operating_flight_number = ""

        dep_time = seg.get("dep_time")
        arr_time = seg.get("arr_time")
        dep_time = time_format_conversion(dep_time, in_strftime_str="%Y%m%d%H%M%S", out_strftime_str="%Y-%m-%d %H:%M:%S")
        arr_time = time_format_conversion(arr_time, in_strftime_str="%Y%m%d%H%M%S", out_strftime_str="%Y-%m-%d %H:%M:%S")

        end_segment = {
            "carrier": seg.get("carrier"),
            "flight_number": flight_number,
            # "dep_air_port": seg.get("dep_air_port"),
            # "arr_air_port": seg.get("arr_air_port"),
            "dep_city_code": seg.get("dep_city_code"),
            "arr_city_code": seg.get("arr_city_code"),
            # "operating_flight_number": operating_flight_number,
            "cabin": seg.get("cabin"),
            "dep_time": dep_time,
            # "arr_time": arr_time,
        }
        end_segments.append(end_segment)

    return {
        "trip_type": 1,
        # "cover_price": adult_price,
        # "cover_tax": adult_tax,
        "bag_amount": pc,
        "bag_weight": kg,
        "max_threshold": max_threshold,
        "segments": end_segments,
        "ret_segments": [],
        "task": task
    }



def sync_policy(payload):
    headers = {
        "Content-Type": "application/json",
    }
    # print(json.dumps(payload, ensure_ascii=False, indent=2))
    response = requests.post(POLICY_URL, headers=headers, json=payload, timeout=30)
    # print(response.text[:1000])
    resp_json = response.json()
    """
    {
        "code": 0,
        "msg": "ok",
        "data": {
            "deleted": 1,
            "created": 7
        }
    }
    """
    # print(json.dumps(resp_json, ensure_ascii=False, indent=2))
    return resp_json



def main():

    logger = XmiLogger("task")

    # 注意  \ufeff 是 UTF-8 的 BOM
    # 所以需要使用 utf-8-sig 编码
    task_list = []
    runner = FlightPriceTaskRunner(logger=logger)

    # 1 读取任务列表
    output_dir = "./keep_0"
    keep_info_path = os.path.join(output_dir, "keep_info.csv")

    with open(keep_info_path, "r", encoding="utf-8-sig") as f:
        reader = csv.DictReader(f)
        for row in reader:
            task_list.append(row)

    # 2 任务列表逻辑处理（多线程）

    policy_list = []
    keep_info_end = []
    max_workers = 5  # 并发线程数，可按需要调整
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(_process_one_task, task, runner): task for task in task_list}
        total = len(futures)
        done = 0
        failed = 0
        for future in as_completed(futures):
            try:
                flight_data = future.result()
                if flight_data is not None:
                    task = flight_data.pop("task")
                    keep_info_end.append(task)
                    policy_list.append(flight_data)
            except Exception as e:
                failed += 1
                task = futures[future]
                # print(f"任务异常 {task}: {e}")
                logger.error(f"任务异常 {task}: {e}")
            finally:
                done += 1
                logger.info(
                    f"进度: {done}/{total}, policy: {len(policy_list)}, keep: {len(keep_info_end)}, failed: {failed}"
                )

    # 3 批量一次性上传政策
    logger.info(f"数据过滤后, 上传政策: {len(policy_list)}")
    # logger.info(f"policy_list: {policy_list}")
    logger.info(f"policy_list: {json.dumps(policy_list, ensure_ascii=False, default=str)[:1000]}")
    if len(policy_list) > 0:
        # 这里批量一次性上传政策
        payload = {"items": policy_list}
        try:
            sync_policy(payload)
            logger.info(f"上传政策成功")
        except Exception as e:
            logger.error(f"上传政策失败: {e}")
            logger.error(f"{traceback.format_exc()}")

    logger.info(f"keep_info_end: {len(keep_info_end)}")
    # 将 keep_info_end 写入到文件csv 文件 嵌套结构要处理  提供下载页面 (历史数据需要保留)
    output_dir = "/home/node04/descending_cabin_files_vj"
    os.makedirs(output_dir, exist_ok=True)
    if keep_info_end:
        out_path = os.path.join(
            output_dir,
            f"keep_info_end_{datetime.now().strftime('%Y%m%d%H%M%S')}.csv",
        )
        with open(out_path, "w", encoding="utf-8-sig") as f:
            writer = csv.DictWriter(f, fieldnames=keep_info_end[0].keys())
            writer.writeheader()
            for task in keep_info_end:
                writer.writerow(task)
        logger.info("keep_info_end 写入完成")
    else:
        logger.warning("keep_info_end 为空，跳过写入CSV")


if __name__ == "__main__":
    main()