#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Miravia Order Report - Main Entry Workflow: 1. Parse CLI arguments, normalize time range 2. Fetch main orders -> sub-orders -> reverse orders via MiraviaClient 3. Aggregate into sub-order granularity DataFrame, export Excel/CSV 4. Render HTML analysis report based on DataFrame """ import argparse import json import os import sys import time import logging from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any, Dict, List, Optional # Allow direct `python scripts/export_orders.py` execution THIS_DIR = Path(__file__).resolve().parent sys.path.insert(0, str(THIS_DIR)) import config from miravia_client import MiraviaClient, MiraviaError from skill_logger import api_tracker, write_skill_log, set_agent_type logger = logging.getLogger("miravia.export") # -------- Data quality tracking -------- class _DataQuality: """Tracks partial-fetch failures so the report can be flagged as INCOMPLETE. Rationale: the Miravia gateway returns transient SYSTEM_ERROR randomly. When a time-window / batch fails after all retries, its records are missing. Instead of silently dropping them (which makes every run produce different numbers), we record the gap here and surface a visible "data incomplete" banner in the report. """ def __init__(self) -> None: self.reset() def reset(self) -> None: self.failed_order_windows: List[Dict[str, str]] = [] self.failed_reverse_windows: List[Dict[str, Any]] = [] self.missing_item_orders: int = 0 def add_order_window(self, after: str, before: str) -> None: self.failed_order_windows.append({"after": after, "before": before}) def add_reverse_window(self, after_ms: Any, before_ms: Any) -> None: self.failed_reverse_windows.append({"after_ms": after_ms, "before_ms": before_ms}) @property def incomplete(self) -> bool: return bool(self.failed_order_windows or self.failed_reverse_windows or self.missing_item_orders) def summary(self) -> str: parts: List[str] = [] if self.failed_order_windows: parts.append(f"{len(self.failed_order_windows)} order time-window(s) failed to fetch") if self.failed_reverse_windows: parts.append(f"{len(self.failed_reverse_windows)} refund/return time-window(s) failed to fetch") if self.missing_item_orders: parts.append(f"{self.missing_item_orders} order(s) missing item details") return "; ".join(parts) def to_dict(self) -> Dict[str, Any]: return { "incomplete": self.incomplete, "summary": self.summary(), "failed_order_windows": list(self.failed_order_windows), "failed_reverse_windows": list(self.failed_reverse_windows), "missing_item_orders": self.missing_item_orders, } # Module-level singleton; reset at the start of each run(). _DATA_QUALITY = _DataQuality() # -------- Time utilities -------- def _parse_dt(s: str, end_of_day: bool = False) -> str: """Accept 'YYYY-MM-DD' or full ISO 8601; return ISO string with timezone offset.""" if not s: raise ValueError("empty datetime") if "T" in s: return s # Assume caller provides valid ISO base = datetime.strptime(s, "%Y-%m-%d") if end_of_day: base = base.replace(hour=23, minute=59, second=59) return base.strftime("%Y-%m-%dT%H:%M:%S") + config.DEFAULT_TZ_OFFSET def _tz() -> timezone: """Derive tzinfo from config.DEFAULT_TZ_OFFSET, supports env var override (e.g., +02:00 / +01:00).""" off = config.DEFAULT_TZ_OFFSET or "+02:00" sign = 1 if off[0] == "+" else -1 hh, mm = off[1:].split(":") return timezone(sign * timedelta(hours=int(hh), minutes=int(mm))) def _parse_tz_offset(off_str: str): """Parse timezone offset string to (sign, hours, minutes). Supports all common formats returned by Miravia API: '+08:00' -> (1, 8, 0) # ISO 8601 with colon '+0800' -> (1, 8, 0) # compact without colon '-05:00' -> (-1, 5, 0) '+02' -> (1, 2, 0) # hour-only """ sign = 1 if off_str[0] == "+" else -1 digits = off_str[1:].replace(":", "") if len(digits) >= 4: hh = int(digits[:2]) mm = int(digits[2:4]) elif len(digits) >= 2: hh = int(digits[:2]) mm = 0 else: hh = int(digits) mm = 0 return sign, hh, mm def _convert_to_local_tz(ts_val) -> str: """Convert API-returned timestamp to configured local timezone (config.DEFAULT_TZ_OFFSET). Miravia API returns timestamps in server timezone (typically +0800 CST). This normalizes them to the configured business timezone (default: +02:00 Madrid CEST). Supports: - ISO strings with colon offset: '2026-06-25T15:30:00+08:00' -> '2026-06-25T09:30:00+02:00' - ISO strings compact offset: '2026-06-25 08:01:40 +0800' -> '2026-06-25T02:01:40+02:00' - Unix timestamps (seconds/ms): 1782356394 -> '2026-06-24T23:59:54+02:00' - Strings without offset: treated as already in configured timezone, offset appended - None / empty: returned as-is """ if ts_val is None or ts_val == "": return ts_val # Handle numeric Unix timestamps (seconds or milliseconds) if isinstance(ts_val, (int, float)): ts_sec = ts_val / 1000.0 if ts_val > 9_999_999_999 else float(ts_val) local_tz = _tz() local_dt = datetime.fromtimestamp(ts_sec, tz=local_tz) return local_dt.strftime("%Y-%m-%dT%H:%M:%S") + config.DEFAULT_TZ_OFFSET s = str(ts_val).strip() # Numeric string (e.g. "1782356394.0" from pandas float) try: num = float(s) if num > 1_000_000_000: # Looks like a Unix timestamp ts_sec = num / 1000.0 if num > 9_999_999_999 else num local_tz = _tz() local_dt = datetime.fromtimestamp(ts_sec, tz=local_tz) return local_dt.strftime("%Y-%m-%dT%H:%M:%S") + config.DEFAULT_TZ_OFFSET except (ValueError, TypeError, OverflowError): pass if len(s) < 19: return s # Too short to be a valid datetime try: # Normalize separator: support both 'T' and space between date and time if "T" not in s[:20]: s = s[:10] + "T" + s[11:19] + s[19:] base = s[:19] off_part = s[19:].strip() dt = datetime.strptime(base, "%Y-%m-%dT%H:%M:%S") if off_part and off_part != "Z" and off_part[0] in "+-": sign, hh, mm = _parse_tz_offset(off_part) src_tz = timezone(sign * timedelta(hours=hh, minutes=mm)) elif off_part == "Z": src_tz = timezone.utc else: # No timezone info — treat as already in configured timezone return base + config.DEFAULT_TZ_OFFSET # If source timezone is already the configured one, skip conversion local_tz = _tz() if src_tz == local_tz: return base + config.DEFAULT_TZ_OFFSET # Convert to configured local timezone local_dt = dt.replace(tzinfo=src_tz).astimezone(local_tz) return local_dt.strftime("%Y-%m-%dT%H:%M:%S") + config.DEFAULT_TZ_OFFSET except (ValueError, IndexError, AttributeError, TypeError): return str(ts_val) # Parsing failed — return original def _last_days_range(n: int) -> (str, str): """Semantics: whole-day range including today going back N days (aligned to config.DEFAULT_TZ_OFFSET timezone). n=1 -> today 00:00:00 ~ today 23:59:59 n=7 -> today-6 00:00:00 ~ today 23:59:59 n<1 -> corrected to 1 with warning. """ if n < 1: print(f"warning: --last-days={n} is invalid, falling back to 1.", file=sys.stderr) n = 1 now = datetime.now(_tz()) end = now.replace(hour=23, minute=59, second=59, microsecond=0) start = (now - timedelta(days=n - 1)).replace(hour=0, minute=0, second=0, microsecond=0) return (start.strftime("%Y-%m-%dT%H:%M:%S") + config.DEFAULT_TZ_OFFSET, end.strftime("%Y-%m-%dT%H:%M:%S") + config.DEFAULT_TZ_OFFSET) def _iso_to_ms(s: str) -> int: """ISO string with timezone offset -> Unix millisecond timestamp.""" # e.g. '2026-04-28T00:00:00+08:00' dt = datetime.strptime(s[:19], "%Y-%m-%dT%H:%M:%S") off = s[19:] or config.DEFAULT_TZ_OFFSET sign = 1 if off[0] == "+" else -1 hh, mm = off[1:].split(":") tz = timezone(sign * timedelta(hours=int(hh), minutes=int(mm))) return int(dt.replace(tzinfo=tz).timestamp() * 1000) _NULL_SENTINELS = (None, "", "null", "null-null", "NULL", "None") def _is_blank(v: Any) -> bool: """Miravia often returns string sentinels like 'null-null' to represent null values.""" return v in _NULL_SENTINELS def _safe_filename(s: str) -> str: """Generate filename-safe ISO string. Preserves timezone offset without truncation. Processing: separates date part from timezone offset, safely replaces each, avoiding incorrect removal of '-' in negative timezones (e.g., -05:00). """ import re # Match trailing timezone offset (+08:00 / -05:00 / Z) m = re.match(r'^(.*?)([+-]\d{2}:\d{2}|Z)?$', s) if m: base, tz = m.group(1), m.group(2) or "" base = base.replace(":", "").replace("T", "_") tz = tz.replace(":", "").replace("+", "p").replace("-", "m") return base + tz return s.replace(":", "").replace("+", "p").replace("-", "m").replace("T", "_") def _resolve_collision(path: Path) -> Path: """If same-named file already exists (same-second re-run), add _1, _2 suffix to avoid overwriting.""" if not path.exists(): return path stem, suffix = path.stem, path.suffix parent = path.parent for i in range(1, 100): cand = parent / f"{stem}_{i}{suffix}" if not cand.exists(): return cand return path # Extreme case, return original path def _cents_to_currency(v): """The reverse order API /reverse/getreverseordersforseller returns refund_amount in smallest currency units (cents), needs /100 to get readable currency amount. Example: API returns 25562 -> 255.62 EUR. /orders/items/get fields like item_price / paid_price / voucher_amount are already in currency units and don't need conversion; this function is exclusively for reverse order amounts. """ if v is None or v == "": return None if isinstance(v, bool): return v # Keep boolean as-is to avoid float(True)=1.0 misconversion try: return round(float(v) / 100.0, 2) except (TypeError, ValueError): return v # -------- data aggregation -------- def _pick_latest_reverse(rev_list: List[Dict[str, Any]]) -> Dict[str, Any]: """Pick the latest reverse order from a list by creation time. Business rule: under current Miravia reverse logic, if an order has one successful refund, a second refund cannot be initiated. So when multiple reverse orders exist for the same item, the earlier ones are all cancelled — only the latest one is meaningful. Selection priority: 1. Sort by apply_time / create_time descending, pick the first. 2. If timestamps are all missing, pick the last element (API returns chronological order, so last = newest). """ if not rev_list: return {} if len(rev_list) == 1: return rev_list[0] def _extract_ts(r: Dict[str, Any]): """Extract comparable timestamp string (ISO or epoch).""" ts = r.get("apply_time") or r.get("create_time") if ts is None or ts == "": return "" return str(ts) # Sort descending by timestamp; empty timestamps sink to bottom sorted_list = sorted(rev_list, key=lambda r: _extract_ts(r), reverse=True) # Pick the one with the latest non-empty timestamp; fallback to last in original list for r in sorted_list: if _extract_ts(r): return r return rev_list[-1] def _flatten(orders: List[Dict[str, Any]], items_by_order: Dict[str, Dict[str, Any]], reverse_by_item: Dict[str, List[Dict[str, Any]]]) -> List[Dict[str, Any]]: """Generate sub-order granularity rows. See reference.md for field mapping. Reverse order deduplication: each item produces exactly 1 row. When a single item has multiple reverse orders, only the LATEST one (by creation time) is used for refund fields. Business rationale: under current Miravia rules, only one successful refund is allowed per item; earlier reverse orders are always cancelled. """ rows: List[Dict[str, Any]] = [] for o in orders: oid = str(o.get("order_id")) addr = o.get("address_shipping") or {} entry = items_by_order.get(oid) or {} items = entry.get("items") or [] # buyer_remark / seller_remark actual source fields come from /orders/items/get response entry entry_buyer_remark = entry.get("buyer_remark") or "" entry_seller_remark = entry.get("seller_remark") or "" first_name = o.get("customer_first_name") or "" last_name = o.get("customer_last_name") or "" buyer_name = (first_name + " " + last_name).strip() or None for it in items: iid = str(it.get("order_item_id")) raw_rev_list = reverse_by_item.get(iid) or [] # Pick only the latest reverse order (dedup: 1 row per item) rev = _pick_latest_reverse(raw_rev_list) if raw_rev_list else {} cancel_initiator_raw = it.get("cancel_return_initiator") cancel_initiator = None if _is_blank(cancel_initiator_raw) else cancel_initiator_raw api_qty = it.get("quantity") try: qty = int(api_qty) if api_qty not in (None, "") else 1 except (TypeError, ValueError): qty = 1 rows.append({ # order-level basics "order_id": oid, "marketplace": o.get("marketplace"), "created_at": _convert_to_local_tz(o.get("created_at")), "items_count": o.get("items_count"), "payment_method": o.get("payment_method"), "buyer_id": o.get("buyer_id"), # prefer real buyer_id; may be None under invite-test token "buyer_name": buyer_name, # concatenated name, no longer mixed with ID "ship_address": ", ".join(filter(None, [ addr.get("address1"), addr.get("address2"), addr.get("address3"), addr.get("city"), addr.get("post_code"), addr.get("country"), ])), "ship_country": addr.get("country"), "warehouse_code": o.get("warehouse_code"), # no longer fallback to delivery_info "buyer_remark": entry_buyer_remark, "seller_remark": entry_seller_remark, # item-level basics "order_item_id": iid, "item_status": it.get("status"), "item_updated_at": _convert_to_local_tz(it.get("updated_at")), # product "product_id": it.get("product_id"), "product_name": it.get("name"), "product_url": it.get("product_main_image"), "sku": it.get("sku"), "shop_sku": it.get("shop_sku"), "quantity": qty, "unit_price": it.get("item_price"), # pricing "buyer_paid": it.get("paid_price"), "discount_total": it.get("voucher_amount"), "discount_platform": it.get("voucher_platform"), "discount_seller": it.get("voucher_seller"), "shipping_paid": it.get("shipping_amount"), "shipping_disc_total": it.get("shipping_amount_voucher"), "shipping_disc_platform": it.get("shipping_service_cost"), "shipping_disc_seller": it.get("shipping_amount_seller"), "currency": it.get("currency") or o.get("currency"), # logistics "shipment_type": it.get("shipment_type"), "tracking_no": it.get("tracking_code"), "package_id": it.get("package_id"), "shipping_provider": it.get("shipping_provider_type") or it.get("shipping_provider"), # cancellation: cancel_return_initiator is "null-null" sentinel when not cancelled "is_canceled": (cancel_initiator is not None) or (it.get("status") in ("canceled", "cancelled")), "cancel_initiator": cancel_initiator, "cancel_reason": it.get("reason") or it.get("reason_detail") or None, # refund "refund_id": rev.get("reverse_order_id") or rev.get("dispute_id"), "refund_apply_at": _convert_to_local_tz(rev.get("apply_time") or rev.get("create_time")), "refund_status": rev.get("status"), # NOTE: reverse order API refund_amount is in smallest currency units (cents), must /100 "refund_amount": _cents_to_currency(rev.get("refund_amount")), "refund_reason": rev.get("reason"), "is_fast_refund": rev.get("fast_refund"), "is_instant_refund": rev.get("instant_refund"), "refund_tracking_no": rev.get("return_tracking_no"), # ---- v4 new fields ---- # address/region "city": addr.get("city") or addr.get("City"), "ship_to": it.get("ship_to") or it.get("shipTo") or addr.get("country"), # channel/tags "channel": it.get("channel") or o.get("channel") or it.get("marketplace") or o.get("marketplace"), "order_flag": it.get("order_flag") or it.get("orderFlag"), # logistics attributes "is_fbl": it.get("is_fbl") or it.get("isFBL") or 0, "is_digital": it.get("is_digital") or it.get("IsDigital") or 0, "is_sof": it.get("is_sof") or it.get("isSOF") or 0, "shipping_fee_original": it.get("shipping_fee_original") or it.get("shippingFeeOriginal"), "shipping_fee_disc_seller": it.get("shipping_fee_discount_seller") or it.get("shippingFeeDiscountSeller"), # product/discount "voucher_seller_lpi": it.get("voucher_seller_lpi") or it.get("voucherSellerLPI"), "variation": it.get("variation") or it.get("Variation"), "returnable": it.get("returnable", True), # reverse order extension (v4) "request_type": rev.get("request_type"), "duty_party": rev.get("duty_party"), "is_not_received": rev.get("is_not_received"), "refund_payment_method": rev.get("refund_payment_method"), "reason_code": rev.get("reason_code"), "ofc_status": rev.get("ofc_status"), "refund_currency": rev.get("refund_currency"), "seller_sku_id": rev.get("seller_sku_id"), "seller_agreed": rev.get("status") in ( "AGREE_CANCEL_ORDER", "SELLER_AGREE_RETURN", "SELLER_AGREE_REFUND", ) if rev.get("status") else None, }) return rows # -------- Excel/CSV export -------- EXPORT_COLUMNS = [ # order-level "order_id", "marketplace", "created_at", "items_count", "payment_method", "buyer_id", "buyer_name", "ship_address", "ship_country", "warehouse_code", "buyer_remark", "seller_remark", # item-level "order_item_id", "item_status", "item_updated_at", # product "product_id", "product_name", "product_url", "sku", "shop_sku", "quantity", "unit_price", # pricing "buyer_paid", "discount_total", "discount_platform", "discount_seller", "shipping_paid", "shipping_disc_total", "shipping_disc_platform", "shipping_disc_seller", "currency", # logistics "shipment_type", "tracking_no", "package_id", "shipping_provider", # cancellation "is_canceled", "cancel_initiator", "cancel_reason", # refund "refund_id", "refund_apply_at", "refund_status", "refund_amount", "refund_reason", "is_fast_refund", "is_instant_refund", "refund_tracking_no", # ---- v4 additions (hidden in merchant Excel, used by HTML report) ---- "city", "ship_to", "channel", "order_flag", "is_fbl", "is_digital", "is_sof", "shipping_fee_original", "shipping_fee_disc_seller", "voucher_seller_lpi", "variation", "returnable", "request_type", "duty_party", "is_not_received", "refund_payment_method", "reason_code", "ofc_status", "refund_currency", "seller_sku_id", "seller_agreed", ] # PRD-defined visible columns for merchant Excel export. # Only these columns are visible; v4 additions are hidden in Excel. PRD_VISIBLE_COLUMNS = [ # order-level basics "order_id", "marketplace", "created_at", "items_count", "payment_method", "buyer_id", "buyer_name", "ship_address", "ship_country", "warehouse_code", "buyer_remark", "seller_remark", # item-level basics "order_item_id", "item_status", "item_updated_at", # product "product_id", "product_name", "product_url", "sku", "shop_sku", "quantity", "unit_price", # pricing "buyer_paid", "discount_total", "discount_platform", "discount_seller", "shipping_fee_original", "shipping_paid", "shipping_disc_total", "shipping_disc_platform", "shipping_disc_seller", "currency", # logistics "shipment_type", "tracking_no", "package_id", "shipping_provider", # cancellation "is_canceled", "cancel_initiator", "cancel_reason", # refund "refund_id", "refund_apply_at", "refund_status", "refund_amount", "refund_reason", "is_fast_refund", "is_instant_refund", "refund_tracking_no", ] # Excel header rename mapping: internal column name -> PRD Excel column name # These renames only apply to Excel export headers, not to internal data. EXCEL_HEADER_RENAME = { "refund_id": "reverse_order_id", "refund_apply_at": "return_order_line_gmt_create", } # These columns are large numeric string IDs/codes; must be written as text to avoid Excel scientific notation or precision loss TEXT_ID_COLUMNS = [ "order_id", "order_item_id", "product_id", "sku", "shop_sku", "package_id", "tracking_no", "refund_id", "refund_tracking_no", "buyer_id", ] def export_dataframe(rows: List[Dict[str, Any]], out_path: Path, fmt: str) -> Path: """Export rows to Excel/CSV. For Excel (xlsx): all EXPORT_COLUMNS are written, but only PRD_VISIBLE_COLUMNS are visible; v4 additions are hidden columns (data preserved for internal use). Column headers are renamed per EXCEL_HEADER_RENAME to match PRD naming. For CSV: only PRD_VISIBLE_COLUMNS are exported (no hidden column support). """ import pandas as pd df = pd.DataFrame(rows, columns=EXPORT_COLUMNS) out_path.parent.mkdir(parents=True, exist_ok=True) # Convert ID columns to pure strings to prevent pandas from inferring as numeric def _to_text_id(v): if v is None: return "" if isinstance(v, float): if pd.isna(v): return "" # int-like float: remove trailing .0 if v.is_integer(): return str(int(v)) return repr(v) return str(v) for col in TEXT_ID_COLUMNS: if col in df.columns: df[col] = df[col].apply(_to_text_id) if fmt == "csv": # CSV: export only PRD visible columns (no hidden column support in CSV) df_csv = df[[c for c in PRD_VISIBLE_COLUMNS if c in df.columns]].copy() df_csv = df_csv.rename(columns=EXCEL_HEADER_RENAME) df_csv.to_csv(out_path, index=False, encoding="utf-8-sig") else: # Excel: write all columns, rename headers, then hide non-PRD columns df_excel = df.rename(columns=EXCEL_HEADER_RENAME) with pd.ExcelWriter(out_path, engine="openpyxl") as writer: df_excel.to_excel(writer, index=False, sheet_name="orders") ws = writer.sheets["orders"] from openpyxl.utils import get_column_letter header = list(df_excel.columns) # Set text format for ID columns (use renamed header names) renamed_text_cols = [ EXCEL_HEADER_RENAME.get(c, c) for c in TEXT_ID_COLUMNS ] for col_name in renamed_text_cols: if col_name not in header: continue col_idx = header.index(col_name) + 1 # 1-based col_letter = get_column_letter(col_idx) # Row 1 is header, set text format starting from row 2 for row_idx in range(2, ws.max_row + 1): cell = ws[f"{col_letter}{row_idx}"] cell.number_format = "@" # Hide non-PRD columns (v4 additions) # Determine visible column names after rename prd_renamed = set( EXCEL_HEADER_RENAME.get(c, c) for c in PRD_VISIBLE_COLUMNS ) for idx, col_name in enumerate(header): col_letter = get_column_letter(idx + 1) if col_name not in prd_renamed: ws.column_dimensions[col_letter].hidden = True return out_path # -------- HTML report -------- def _jinja_env(): from jinja2 import Environment, FileSystemLoader return Environment( loader=FileSystemLoader(str(THIS_DIR.parent / "templates")), autoescape=True, ) def render_v4_report(rows: List[Dict[str, Any]], out_path: Path, range_label: str, lang: str = "en") -> Path: """v4 unified dashboard: merges business analysis + refund report into one interactive dashboard. Data injection: Python pre-computes all metrics, injected as window.DATA = {...} into HTML; frontend JS handles time range filtering and chart rendering. """ import report_v4 return report_v4.render( rows=rows, out_path=out_path, range_label=range_label, jinja_env=_jinja_env(), export_columns=EXPORT_COLUMNS, default_max_orders=config.DEFAULT_MAX_ORDERS, lang=lang, ) # -------- inspect scenario: single-order text diagnostics -------- def _fmt(v: Any, default: str = "-") -> str: if v is None or _is_blank(v): return default return str(v) def _format_inspect_block(oid: str, entry: Optional[Dict[str, Any]], rev_list: List[Dict[str, Any]]) -> str: """Render diagnostic text for a single order (can be relayed directly to seller by Agent).""" lines: List[str] = [] lines.append(f"========== Order {oid} ==========") if not entry: lines.append(" \u26a0 Order not found in /orders/items/get response. Please verify the order ID belongs to the currently authorized store.") lines.append("=============================================") return "\n".join(lines) items = entry.get("order_items") if items is None: items = entry.get("items") if items is None: items = [] lines.append(f" Sub-orders : {len(items)}") for it in items: iid = _fmt(it.get("order_item_id")) status = _fmt(it.get("status")) cur = _fmt(it.get("currency"), "") price = _fmt(it.get("paid_price")) name = _fmt(it.get("name")) lines.append(f" - item={iid} [{status}] {cur} {price} {name}") track = it.get("tracking_code") provider = it.get("shipping_provider_type") or it.get("shipping_provider") ship_type = it.get("shipment_type") if track or provider or ship_type: lines.append( f" Logistics : tracking={_fmt(track)} provider={_fmt(provider)} " f"shipment_type={_fmt(ship_type)} package_id={_fmt(it.get('package_id'))}" ) cancel_initiator = it.get("cancel_return_initiator") if cancel_initiator and not _is_blank(cancel_initiator): lines.append( f" Cancel : initiator={cancel_initiator} " f"reason={_fmt(it.get('reason') or it.get('reason_detail'))}" ) if not rev_list: lines.append(" Refund : no reverse orders") else: lines.append(f" Refund : {len(rev_list)} reverse order(s)") for r in rev_list: rid = _fmt(r.get("reverse_order_id") or r.get("dispute_id")) req_type = _fmt(r.get("request_type")) refund_pm = _fmt(r.get("refund_payment_method")) ofc = _fmt(r.get("ofc_status")) inner_lines = r.get("reverse_order_lines") or [] if not inner_lines: lines.append( f" - reverse_order_id={rid} type={req_type} ofc={ofc} payment={refund_pm}" ) continue for ln in inner_lines: status = _fmt(ln.get("reverse_status")) amt = ln.get("refund_amount") cur = _fmt(ln.get("refund_currency") or r.get("refund_currency"), "") reason = _fmt(ln.get("reason_text")) rtrack = _fmt(ln.get("tracking_number")) lines.append(f" - reverse_order_id={rid} type={req_type} status={status}") lines.append(f" Refund reason : {reason}") lines.append(f" Refund amount : {_fmt(amt)} {cur} (payment_method={refund_pm})") lines.append(f" Return status : ofc={ofc} return_tracking={rtrack}") fast = ln.get("is_fast_refund") instant = ln.get("is_instant_refund") if fast is not None or instant is not None: lines.append( f" Fast refund : fast_refund={_fmt(fast)} instant_refund={_fmt(instant)}" ) lines.append("=============================================") return "\n".join(lines) def run_inspect(args: argparse.Namespace, client: MiraviaClient) -> int: """inspect scenario: query order status/logistics/cancel/refund by order IDs, output structured text to stdout. No files written.""" raw = (args.order_ids or "").strip() if not raw: print("error: --scenario inspect requires --order-ids " "(comma-separated trade order ids)", file=sys.stderr) return 1 order_ids = [s.strip() for s in raw.split(",") if s.strip()] if not order_ids: print("error: --order-ids is empty after parsing", file=sys.stderr) return 1 print(f"scenario : inspect (order_ids={len(order_ids)})") # 1. Fetch item details/status/logistics (official API supports batch) items_resp = client.get_multi_order_items(order_ids) items_by_order: Dict[str, Dict[str, Any]] = {} for entry in items_resp: oid = str(entry.get("order_id")) items_by_order[oid] = entry # 2. Reverse orders must be fetched per-order: /reverse API trade_order_id is single-value filter reverse_by_order: Dict[str, List[Dict[str, Any]]] = {} for oid in order_ids: try: rev_list = client.get_reverse_orders( trade_order_id=oid, page_size=100, max_pages=3, ) except (MiraviaError, ValueError) as e: logger.warning("fetch reverse for %s failed: %s", oid, e) rev_list = [] reverse_by_order[oid] = rev_list # 3. Render output print() for oid in order_ids: block = _format_inspect_block( oid, items_by_order.get(oid), reverse_by_order.get(oid, []), ) print(block) print() return 0 # -------- time-window auto-splitting -------- def _split_time_range(after: str, before: str) -> List[tuple]: """Split [after, before] into weekly (week_start, week_end) tuples. Each window is 7 days (last window may be shorter). Compared to daily splitting (365 windows), weekly splitting only needs ~52 windows, greatly reducing API call count. """ from datetime import datetime, timedelta fmt = "%Y-%m-%dT%H:%M:%S" tz_off = config.DEFAULT_TZ_OFFSET start_dt = datetime.strptime(after[:19], fmt) end_dt = datetime.strptime(before[:19], fmt) windows = [] cursor = start_dt while cursor <= end_dt: # Each window spans 7 days; window end is cursor + 6 days at 23:59:59 window_end = (cursor + timedelta(days=6)).replace(hour=23, minute=59, second=59) if window_end > end_dt: window_end = end_dt windows.append(( cursor.strftime(fmt) + tz_off, window_end.strftime(fmt) + tz_off, )) cursor = (cursor + timedelta(days=7)).replace(hour=0, minute=0, second=0) return windows def _split_week_to_days(w_after: str, w_before: str) -> List[tuple]: """Split a weekly window into daily sub-windows (for adaptive degradation).""" from datetime import datetime, timedelta fmt = "%Y-%m-%dT%H:%M:%S" tz_off = config.DEFAULT_TZ_OFFSET start_dt = datetime.strptime(w_after[:19], fmt) end_dt = datetime.strptime(w_before[:19], fmt) days = [] cursor = start_dt while cursor <= end_dt: day_end = cursor.replace(hour=23, minute=59, second=59) if day_end > end_dt: day_end = end_dt days.append(( cursor.strftime(fmt) + tz_off, day_end.strftime(fmt) + tz_off, )) cursor = (cursor + timedelta(days=1)).replace(hour=0, minute=0, second=0) return days def _fetch_orders_auto_split( client: "MiraviaClient", after: str, before: str, api_status: Optional[str], max_orders: int, channel: Optional[str] = None, ship_to: Optional[str] = None, order_numbers: Optional[str] = None, ) -> List[Dict[str, Any]]: """Fetch orders with automatic time-window splitting (supports concurrency). Strategy: 1. First attempt to fetch within the full time range (limit: ORDERS_PER_WINDOW). 2. If full window data exceeds ORDERS_PER_WINDOW, split by day. 3. When total > CONCURRENT_THRESHOLD, enable thread pool concurrency. 4. Stop when total reaches max_orders. """ from concurrent.futures import ThreadPoolExecutor, as_completed per_window = config.ORDERS_PER_WINDOW # 5000 (API returns empty at offset>=5000) # Probe total count (first page gives countTotal; internal use only for split decision) first_page = client.get_orders_page( created_after=after, created_before=before, status=api_status, limit=1, offset=0, channel=channel, ship_to=ship_to, order_numbers=order_numbers, ) total_in_range = int(first_page.get("countTotal", 0)) # NOTE: countTotal is an API estimate; may differ from actual fetchable count; used only for internal split decision print("progress: fetching orders...", flush=True) if total_in_range == 0: return [] # Single window sufficient: fetch directly if total_in_range <= per_window: orders = list(client.iter_orders( created_after=after, created_before=before, status=api_status, max_orders=min(total_in_range, max_orders), quiet=False, channel=channel, ship_to=ship_to, order_numbers=order_numbers, )) return orders # Need splitting windows = _split_time_range(after, before) use_concurrent = (total_in_range >= config.CONCURRENT_THRESHOLD and len(windows) > 10) workers = config.CONCURRENT_WORKERS if use_concurrent else 1 mode_label = f"concurrent({workers} workers)" if use_concurrent else "sequential" print( f"progress: splitting into {len(windows)} time windows" f" ({per_window} orders per window max, {mode_label})", flush=True, ) all_orders: List[Dict[str, Any]] = [] seen_ids: set = set() def _fetch_window(w_after: str, w_before: str) -> List[Dict[str, Any]]: """Worker: fetch orders for a single window. Adaptive degradation: probes window order count; if it exceeds ADAPTIVE_DEGRADE_THRESHOLD, automatically splits the week into daily sub-windows. """ # Probe window order count probe = client.get_orders_page( created_after=w_after, created_before=w_before, status=api_status, limit=1, offset=0, channel=channel, ship_to=ship_to, order_numbers=order_numbers, ) window_total = int(probe.get("countTotal", 0)) # Adaptive degradation: split to days if threshold exceeded if window_total > config.ADAPTIVE_DEGRADE_THRESHOLD: logger.info("adaptive degrade: window %s~%s has %d orders (>%d), splitting to days", w_after[:10], w_before[:10], window_total, config.ADAPTIVE_DEGRADE_THRESHOLD) day_windows = _split_week_to_days(w_after, w_before) results = [] for d_after, d_before in day_windows: day_orders = list(client.iter_orders( created_after=d_after, created_before=d_before, status=api_status, max_orders=per_window, quiet=True, channel=channel, ship_to=ship_to, order_numbers=order_numbers, )) results.extend(day_orders) return results # Normal fetch if window_total == 0: return [] return list(client.iter_orders( created_after=w_after, created_before=w_before, status=api_status, max_orders=per_window, quiet=True, channel=channel, ship_to=ship_to, order_numbers=order_numbers, )) if not use_concurrent: # Sequential mode (small order volume, preserves original behavior) failed_windows: List[tuple] = [] for idx, (w_after, w_before) in enumerate(windows, 1): if len(all_orders) >= max_orders: break try: batch = _fetch_window(w_after, w_before) except Exception as e: # noqa: BLE001 - collect for retry instead of aborting run logger.warning("window %d/%d (%s~%s) failed: %s", idx, len(windows), w_after[:10], w_before[:10], e) failed_windows.append((w_after, w_before)) continue for o in batch: oid = str(o.get("order_id")) if oid not in seen_ids: seen_ids.add(oid) all_orders.append(o) if len(all_orders) >= max_orders: break print( f"progress: fetched {len(all_orders)} orders" f" (window {idx}/{len(windows)})", flush=True, ) # Window-level second retry for failed windows. if failed_windows and len(all_orders) < max_orders: print(f"progress: retrying {len(failed_windows)} failed time window(s)...", flush=True) still_failed: List[tuple] = [] for w_after, w_before in failed_windows: try: batch = _fetch_window(w_after, w_before) except Exception as e: # noqa: BLE001 logger.warning("window retry (%s~%s) failed: %s", w_after[:10], w_before[:10], e) still_failed.append((w_after, w_before)) continue for o in batch: oid = str(o.get("order_id")) if oid not in seen_ids: seen_ids.add(oid) all_orders.append(o) failed_windows = still_failed # Record unrecoverable windows so the report can flag INCOMPLETE data. for w_after, w_before in failed_windows: _DATA_QUALITY.add_order_window(w_after, w_before) else: # Concurrent mode: use thread pool for batch fetching. def _run_windows(win_list, workers_n): """Fetch a list of windows concurrently; return (results_map, failed_list).""" results_map: Dict[tuple, List[Dict[str, Any]]] = {} failed_list: List[tuple] = [] done = 0 with ThreadPoolExecutor(max_workers=workers_n) as pool: futures = { pool.submit(_fetch_window, wa, wb): (wa, wb) for (wa, wb) in win_list } for future in as_completed(futures): done += 1 wa, wb = futures[future] try: results_map[(wa, wb)] = future.result() except Exception as e: # noqa: BLE001 - collect for retry logger.warning("window %s~%s failed: %s", wa[:10], wb[:10], e) failed_list.append((wa, wb)) if done % 5 == 0 or done == len(futures): print( f"progress: fetched {len(results_map)} of {len(win_list)} window(s)", flush=True, ) return results_map, failed_list results_map, failed_windows = _run_windows(windows, workers) # Window-level second retry with reduced concurrency to lower QPS pressure. if failed_windows: print(f"progress: retrying {len(failed_windows)} failed time window(s)...", flush=True) retry_workers = max(1, workers // 2) retry_map, failed_windows = _run_windows(failed_windows, retry_workers) results_map.update(retry_map) # Record unrecoverable windows so the report can flag INCOMPLETE data. for w_after, w_before in failed_windows: _DATA_QUALITY.add_order_window(w_after, w_before) # Assemble results in window order for deterministic de-dup. for (w_after, w_before) in windows: batch = results_map.get((w_after, w_before)) if not batch: continue for o in batch: oid = str(o.get("order_id")) if oid not in seen_ids: seen_ids.add(oid) all_orders.append(o) # Truncate in concurrent mode if len(all_orders) > max_orders: all_orders = all_orders[:max_orders] return all_orders # -------- reverse order time-window auto-splitting -------- def _split_reverse_time_range_monthly(after_ms: int, before_ms: int) -> List[tuple]: """Split millisecond timestamp range into monthly (start_ms, end_ms) tuples. Each window aligns to natural month boundaries (whole-day aligned), ensuring reverse order count per window doesn't exceed server deep-page limit (max_pages=30 x page_size=100 = 3000). """ from datetime import datetime, timedelta, timezone tz_off = config.DEFAULT_TZ_OFFSET or "+02:00" sign = 1 if tz_off[0] == "+" else -1 hh, mm = tz_off[1:].split(":") tz = timezone(sign * timedelta(hours=int(hh), minutes=int(mm))) start_dt = datetime.fromtimestamp(after_ms / 1000, tz=tz) end_dt = datetime.fromtimestamp(before_ms / 1000, tz=tz) windows = [] cursor = start_dt.replace(day=1, hour=0, minute=0, second=0, microsecond=0) while cursor <= end_dt: # Month end: next month 1st - 1 second if cursor.month == 12: next_month = cursor.replace(year=cursor.year + 1, month=1) else: next_month = cursor.replace(month=cursor.month + 1) month_end = next_month - timedelta(seconds=1) # Clip to actual query range w_start = max(cursor, start_dt) w_end = min(month_end, end_dt) w_start_ms = int(w_start.timestamp() * 1000) w_end_ms = int(w_end.timestamp() * 1000) if w_start_ms <= w_end_ms: windows.append((w_start_ms, w_end_ms)) cursor = next_month return windows def _fetch_reverse_auto_split( client: "MiraviaClient", after_ms: int, before_ms: int, ) -> List[Dict[str, Any]]: """Fetch reverse orders with automatic monthly splitting to avoid server deep-page limit (supports concurrency). Strategy: 1. If time range <= 30 days, fetch directly (single month unlikely to exceed 3000 records). 2. If time range > 30 days, split by natural month, fetch concurrently, deduplicate across windows. Compared to previous single full-range fetch, this function ensures no reverse order records are lost. """ from concurrent.futures import ThreadPoolExecutor, as_completed days = (before_ms - after_ms) / (1000 * 86400) # Short time range: fetch directly, but check for potential truncation if days <= 30: try: results = client.get_reverse_orders( created_after_ms=after_ms, created_before_ms=before_ms, ) except (MiraviaError, ValueError) as e: # Second retry before giving up. logger.warning("reverse short-range fetch failed: %s; retrying once", e) try: results = client.get_reverse_orders( created_after_ms=after_ms, created_before_ms=before_ms, ) except (MiraviaError, ValueError) as e2: logger.warning("reverse short-range retry failed: %s", e2) _DATA_QUALITY.add_reverse_window(after_ms, before_ms) return [] # If we got near the max_pages * page_size cap (3000), data may have been # silently truncated. Fall back to weekly splitting to ensure completeness. max_expected = 30 * 100 # default max_pages * page_size if len(results) >= max_expected * 0.95: logger.warning( "reverse short-range may have truncated data (%d records, near cap %d), " "falling back to weekly splitting", len(results), max_expected, ) from datetime import datetime, timedelta tz = _tz() start_dt = datetime.fromtimestamp(after_ms / 1000, tz=tz) end_dt = datetime.fromtimestamp(before_ms / 1000, tz=tz) all_results: List[Dict[str, Any]] = [] seen_ids: set = set() cursor = start_dt while cursor <= end_dt: w_end = min(cursor + timedelta(days=7), end_dt) w_after_ms = int(cursor.timestamp() * 1000) w_before_ms = int(w_end.timestamp() * 1000) batch = None for _attempt in range(2): # initial + one window-level retry try: batch = client.get_reverse_orders( created_after_ms=w_after_ms, created_before_ms=w_before_ms, ) break except (MiraviaError, ValueError) as e: logger.warning("reverse weekly window failed (attempt %d): %s", _attempt + 1, e) if batch is None: _DATA_QUALITY.add_reverse_window(w_after_ms, w_before_ms) batch = [] for r in batch: rid = (r.get("reverse_order_id") or r.get("dispute_id") or id(r)) if rid not in seen_ids: seen_ids.add(rid) all_results.append(r) cursor = w_end + timedelta(seconds=1) return all_results return results # Long time range: split by month windows = _split_reverse_time_range_monthly(after_ms, before_ms) use_concurrent = len(windows) > 3 workers = config.CONCURRENT_WORKERS if use_concurrent else 1 logger.info("reverse auto-split: %d monthly windows for %.0f-day range (%s)", len(windows), days, f"concurrent({workers})" if use_concurrent else "sequential") all_results: List[Dict[str, Any]] = [] seen_ids: set = set() def _fetch_month(w_after: int, w_before: int) -> List[Dict[str, Any]]: return client.get_reverse_orders( created_after_ms=w_after, created_before_ms=w_before, ) if not use_concurrent: failed_rev_windows: List[tuple] = [] for idx, (w_after, w_before) in enumerate(windows, 1): try: batch = _fetch_month(w_after, w_before) except (MiraviaError, ValueError) as e: logger.warning("reverse window %d/%d failed: %s", idx, len(windows), e) failed_rev_windows.append((w_after, w_before)) continue for r in batch: rid = (r.get("reverse_order_id") or r.get("dispute_id") or id(r)) if rid not in seen_ids: seen_ids.add(rid) all_results.append(r) # Window-level second retry. if failed_rev_windows: logger.info("reverse retry: %d failed window(s)", len(failed_rev_windows)) still_failed: List[tuple] = [] for w_after, w_before in failed_rev_windows: try: batch = _fetch_month(w_after, w_before) except (MiraviaError, ValueError) as e: logger.warning("reverse window retry failed: %s", e) still_failed.append((w_after, w_before)) continue for r in batch: rid = (r.get("reverse_order_id") or r.get("dispute_id") or id(r)) if rid not in seen_ids: seen_ids.add(rid) all_results.append(r) failed_rev_windows = still_failed for w_after, w_before in failed_rev_windows: _DATA_QUALITY.add_reverse_window(w_after, w_before) else: def _run_rev_windows(win_list, workers_n): """Fetch reverse windows concurrently; return (results_list, failed_list).""" local_results: List[Dict[str, Any]] = [] failed_list: List[tuple] = [] with ThreadPoolExecutor(max_workers=workers_n) as pool: futures = { pool.submit(_fetch_month, wa, wb): (wa, wb) for (wa, wb) in win_list } for future in as_completed(futures): wa, wb = futures[future] try: local_results.extend(future.result()) except Exception as e: # noqa: BLE001 - collect for retry logger.warning("reverse window %s~%s failed: %s", wa, wb, e) failed_list.append((wa, wb)) return local_results, failed_list batch_results, failed_rev_windows = _run_rev_windows(windows, workers) # Window-level second retry with reduced concurrency. if failed_rev_windows: logger.info("reverse retry: %d failed window(s)", len(failed_rev_windows)) retry_workers = max(1, workers // 2) retry_results, failed_rev_windows = _run_rev_windows(failed_rev_windows, retry_workers) batch_results.extend(retry_results) for r in batch_results: rid = (r.get("reverse_order_id") or r.get("dispute_id") or id(r)) if rid not in seen_ids: seen_ids.add(rid) all_results.append(r) for w_after, w_before in failed_rev_windows: _DATA_QUALITY.add_reverse_window(w_after, w_before) logger.info("reverse auto-split done: %d records from %d windows", len(all_results), len(windows)) return all_results # -------- token pre-check -------- def _check_token_expiry() -> Optional[str]: """Check if token_expires_at in credentials.json has expired. Returns None if token is valid (or cannot be determined), returns error message if expired. """ if not os.path.exists(config.CRED_FILE): return None try: with open(config.CRED_FILE, "r", encoding="utf-8") as f: data = json.load(f) expires_str = data.get("token_expires_at", "") if not expires_str: return None # Supports both "2026-06-24 18:27" and "2026-06-24T18:27:00" formats for fmt in ("%Y-%m-%d %H:%M", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%d %H:%M:%S"): try: expires_dt = datetime.strptime(expires_str, fmt) break except ValueError: continue else: return None # Cannot parse, skip check now = datetime.now() if now > expires_dt: return ( f"Token expired (expires_at={expires_str}, now={now.strftime('%Y-%m-%d %H:%M:%S')}).\n" f"Please re-authorize: python scripts/auth_cli.py login" ) # Near-expiry warning (less than 2 minutes remaining) remaining = (expires_dt - now).total_seconds() if remaining < 120: print( f"warning: token will expire in {int(remaining)}s, consider re-authorizing soon.", file=sys.stderr, ) return None except (OSError, json.JSONDecodeError, Exception): return None # Read failure should not block the flow def _preflight_token_check(client: MiraviaClient) -> Optional[str]: """Pre-flight check: first check local expiry time, then verify token validity via ping. Returns None if token is valid, returns error message if token is invalid. """ # 1. Local quick check: token_expires_at expiry_err = _check_token_expiry() if expiry_err: return expiry_err # 2. Remote verification: lightweight ping try: client.ping() return None except MiraviaError as e: if "IllegalAccessToken" in str(e.code) or "AccessTokenExpired" in str(e.code): return ( f"Token verification failed (API returned {e.code}): {e.msg}\n" f"Please re-authorize: python scripts/auth_cli.py login" ) # Other error types (e.g., network issues) should not block logger.warning("preflight ping failed with non-auth error: %s", e) return None except Exception as e: logger.warning("preflight ping unexpected error: %s", e) return None # -------- main flow -------- def run(args: argparse.Namespace) -> int: # Reset API call tracker, start new timing round api_tracker.reset() # Reset data-quality tracker so each run reports its own completeness state. _DATA_QUALITY.reset() _start_ms = int(time.time() * 1000) # Determine action: ping / inspect / scenario(arg) / auto _action = "ping" if args.ping else (args.scenario or "auto") client = MiraviaClient(action=_action) if args.ping: try: client.ping() print("ping OK") _duration = int(time.time() * 1000) - _start_ms write_skill_log(action="ping", status="success", duration_ms=_duration, data_count=0) return 0 except MiraviaError as e: print(f"ping FAIL: {e}", file=sys.stderr) _duration = int(time.time() * 1000) - _start_ms write_skill_log(action="ping", status="fail", duration_ms=_duration, error_code=e.code) return 1 # \u2605 Pre-flight token validation: confirm token is valid before any data fetching # (including inspect scenario, so expired token gives friendly re-auth prompt) token_err = _preflight_token_check(client) if token_err: print(f"error: {token_err}", file=sys.stderr) _duration = int(time.time() * 1000) - _start_ms write_skill_log( action=(_action if _action != "ping" else "auto"), status="fail", duration_ms=_duration, error_code="TOKEN_EXPIRED", remark="pre-flight token validation failed") return 1 # inspect scenario uses single-order direct query path if (args.scenario or "").lower() == "inspect": rc = run_inspect(args, client) _duration = int(time.time() * 1000) - _start_ms write_skill_log( action="inspect", status="success" if rc == 0 else "fail", duration_ms=_duration, data_count=0, remark=f"order_ids={args.order_ids or ''}") return rc # When --order-ids is specified (non-inspect), skip time range requirement (fast path) if args.order_ids and (args.scenario or "").lower() != "inspect": after, before = None, None elif args.last_days is not None: after, before = _last_days_range(args.last_days) elif args.created_after and args.created_before: after = _parse_dt(args.created_after) before = _parse_dt(args.created_before, end_of_day=True) else: print( "error: time range required. Use --last-days N or --created-after/--created-before.\n" " (Core Rule #2: default time range is prohibited)", file=sys.stderr, ) return 1 if after and before: print(f"range : {after} ~ {before}") api_status = _resolve_api_status(args.status) api_channel = getattr(args, 'channel', None) api_ship_to = getattr(args, 'ship_to', None) api_order_numbers = args.order_ids # ★ Fast path: when --order-ids is specified with few orders (≤10), # use /orders/items/get + /reverse/getreverseordersforseller directly # (because /orders/get does NOT support orderNumbers filtering). _fast_path_ids = ( [x.strip() for x in api_order_numbers.split(",") if x.strip()] if api_order_numbers else [] ) _use_fast_path = len(_fast_path_ids) > 0 and len(_fast_path_ids) <= 10 if _use_fast_path: # ───── Fast path: /order/get + /orders/items/get + /reverse per order ───── # /orders/get (plural) does NOT support order_id filtering; # /order/get (singular) fetches complete single-order info by order_id; # /orders/items/get and /reverse DO support order_id filtering. print("progress: fast-path query by order ID(s)...", flush=True) # Step 1: /orders/items/get with order_ids (guaranteed to filter server-side) items_resp = client.get_multi_order_items(_fast_path_ids) if not items_resp: print( f"no_data : No orders found for order ID(s): {api_order_numbers}. " f"Please verify the order ID belongs to the currently authorized store.", flush=True, ) _duration = int(time.time() * 1000) - _start_ms write_skill_log( action=(args.scenario or "auto"), status="no_data", duration_ms=_duration, data_count=0, remark=f"fast_path order_ids={api_order_numbers} not_found") return 3 # Step 2: /order/get for each order to get complete order-level fields print("progress: fetching full order details...", flush=True) full_order_map: Dict[str, Dict[str, Any]] = {} for fid in _fast_path_ids: full_order = client.get_order(fid) if full_order: full_order_map[str(full_order.get("order_id") or fid)] = full_order # Build items_by_order + orders list using /order/get data (fallback to items response) items_by_order: Dict[str, Dict[str, Any]] = {} orders: List[Dict[str, Any]] = [] order_ids: List[str] = [] for entry in items_resp: oid = str(entry.get("order_id")) order_ids.append(oid) # Get full order data from /order/get; use as primary source full = full_order_map.get(oid) or {} items_by_order[oid] = { "buyer_remark": full.get("buyer_remark_text") or entry.get("buyer_remark_text") or "", "seller_remark": full.get("seller_remark_text") or entry.get("seller_remark_text") or "", "items": entry.get("order_items") if entry.get("order_items") is not None else (entry.get("items") if entry.get("items") is not None else []), } # Order entry — prefer /order/get fields, fallback to /orders/items/get orders.append({ "order_id": oid, "marketplace": full.get("marketplace") or entry.get("marketplace"), "created_at": full.get("created_at") or entry.get("created_at"), "items_count": full.get("items_count") or entry.get("items_count"), "payment_method": full.get("payment_method") or entry.get("payment_method"), "buyer_id": full.get("buyer_id") or entry.get("buyer_id"), "customer_first_name": full.get("customer_first_name") or entry.get("customer_first_name"), "customer_last_name": full.get("customer_last_name") or entry.get("customer_last_name"), "address_shipping": full.get("address_shipping") or entry.get("address_shipping"), "warehouse_code": full.get("warehouse_code") or entry.get("warehouse_code"), "currency": full.get("currency") or entry.get("currency"), }) # Step 3: /reverse per order using trade_order_id print("progress: fetching refund/return data...", flush=True) reverse_list: List[Dict[str, Any]] = [] for oid in order_ids: try: rev_batch = client.get_reverse_orders( trade_order_id=oid, page_size=100, max_pages=3, ) reverse_list.extend(rev_batch) except (MiraviaError, ValueError) as e: logger.warning("fast-path reverse for order %s failed: %s", oid, e) else: # ───── Normal path: /orders/get → /orders/items/get → /reverse ───── # 1. Fetch orders (auto time-window splitting) orders = _fetch_orders_auto_split( client, after, before, api_status=api_status, max_orders=args.max_orders, channel=api_channel, ship_to=api_ship_to, ) if not orders: # ★ Distinguish "truly no data" from "empty result due to API failures" total_calls = api_tracker.call_count success_calls = api_tracker.success_count if total_calls > 1 and success_calls <= 1: print( f"error: API calls mostly failed ({success_calls}/{total_calls} succeeded), " f"likely token expired or network error. Aborting report generation.\n" f"Please re-authorize: python scripts/auth_cli.py login", file=sys.stderr, ) _duration = int(time.time() * 1000) - _start_ms write_skill_log( action=(args.scenario or "auto"), status="fail", duration_ms=_duration, data_count=0, error_code="API_MASS_FAILURE", remark=f"range={after}~{before} api_success={success_calls}/{total_calls}") return 1 print( f"no_data : No orders found for the specified conditions " f"(range={after} ~ {before}, status={args.status or 'all'}, " f"channel={api_channel or 'all'}, ship_to={api_ship_to or 'all'}).", flush=True, ) _duration = int(time.time() * 1000) - _start_ms write_skill_log( action=(args.scenario or "auto"), status="no_data", duration_ms=_duration, data_count=0, remark=f"range={after}~{before} status={args.status} channel={api_channel} ship_to={api_ship_to}") return 3 # 2. Fetch order items print("progress: fetching order items...", flush=True) order_ids = [str(o.get("order_id")) for o in orders if o.get("order_id")] use_items_concurrent = len(order_ids) > config.CONCURRENT_THRESHOLD items_resp = client.get_multi_order_items(order_ids, concurrent=use_items_concurrent) items_by_order: Dict[str, Dict[str, Any]] = {} for entry in items_resp: oid = str(entry.get("order_id")) items_by_order[oid] = { "buyer_remark": entry.get("buyer_remark_text") or "", "seller_remark": entry.get("seller_remark_text") or "", "items": entry.get("order_items") if entry.get("order_items") is not None else (entry.get("items") if entry.get("items") is not None else []), } # 2b. Item batches can be silently dropped on transient API failure. # Detect orders missing item details and retry them once; record any gap. missing_ids = [oid for oid in order_ids if oid not in items_by_order] if missing_ids: print(f"progress: retrying {len(missing_ids)} order(s) missing item details...", flush=True) try: retry_resp = client.get_multi_order_items(missing_ids, concurrent=False) except Exception as e: # noqa: BLE001 logger.warning("item detail retry failed: %s", e) retry_resp = [] for entry in retry_resp: oid = str(entry.get("order_id")) items_by_order[oid] = { "buyer_remark": entry.get("buyer_remark_text") or "", "seller_remark": entry.get("seller_remark_text") or "", "items": entry.get("order_items") if entry.get("order_items") is not None else (entry.get("items") if entry.get("items") is not None else []), } still_missing = [oid for oid in order_ids if oid not in items_by_order] if still_missing: _DATA_QUALITY.missing_item_orders += len(still_missing) logger.warning("%d order(s) still missing item details after retry", len(still_missing)) # 3. Fetch reverse orders (full time-range scan) print("progress: fetching refund/return data...", flush=True) after_ms = _iso_to_ms(after) before_ms = _iso_to_ms(before) reverse_list = _fetch_reverse_auto_split( client, after_ms=after_ms, before_ms=before_ms, ) reverse_by_item: Dict[str, List[Dict[str, Any]]] = {} for r in reverse_list: rev_id = r.get("reverse_order_id") or r.get("dispute_id") lines = r.get("reverse_order_lines") or [] if lines: for line in lines: oil = (line.get("trade_order_line_id") or line.get("marketplace_trade_order_line_id") or line.get("reverse_order_line_id")) if not oil: continue rev_entry = { "reverse_order_id": rev_id, "apply_time": line.get("return_order_line_gmt_create") or r.get("apply_time"), "create_time": line.get("trade_order_gmt_create") or r.get("create_time"), "status": line.get("reverse_status") or r.get("status"), "refund_amount": line.get("refund_amount"), "reason": line.get("reason_text") or r.get("reason"), "fast_refund": line.get("is_fast_refund"), "instant_refund": line.get("is_instant_refund"), "return_tracking_no": line.get("return_tracking_no") or r.get("return_tracking_no"), # v4 addition: pass through from order/item level "request_type": r.get("request_type"), # ★ order-level pass-through "duty_party": line.get("duty_party") or line.get("dutyParty"), "is_not_received": line.get("is_not_received") or line.get("isNotReceived"), "refund_payment_method": line.get("refund_payment_method") or line.get("refundPaymentMethod"), "reason_code": line.get("reason_code") or line.get("reasonCode"), "ofc_status": line.get("ofc_status") or line.get("ofcStatus"), "refund_currency": line.get("refund_currency") or line.get("refundCurrency"), "seller_sku_id": line.get("seller_sku_id") or line.get("sellerSkuId"), } reverse_by_item.setdefault(str(oil), []).append(rev_entry) else: for k in ("order_item_id", "trade_order_line_id", "sub_order_id"): v = r.get(k) if v: reverse_by_item.setdefault(str(v), []).append( {**r, "reverse_order_id": rev_id} ) break # 4. Aggregate rows = _flatten(orders, items_by_order, reverse_by_item) main_order_count = len(set(str(o.get("order_id")) for o in orders if o.get("order_id"))) total_sub_orders = len(rows) print(f"progress: {main_order_count} main orders, {total_sub_orders} sub-order rows total") # Check if exceeding limit hit_cap = total_sub_orders >= args.max_orders if hit_cap: # Truncate to limit rows = rows[:args.max_orders] print( f"warning: data exceeds export limit ({args.max_orders} sub-orders). " f"Exported first {args.max_orders} records. " f"Please narrow the time range and retry for complete data.", file=sys.stderr, ) # 5. Determine outputs by scenario out_dir = Path(args.output).expanduser().resolve() if after and before: fname_range = f"{_safe_filename(after)}_{_safe_filename(before)}" range_label = f"{after} ~ {before}" else: # Fast path (--order-ids without time range) id_list = [x.strip() for x in args.order_ids.split(',')] fname_range = f"{len(id_list)}orders" range_label = f"order_ids: {args.order_ids}" ts = time.strftime("%Y%m%d%H%M%S") scenario, do_excel, do_report = _resolve_outputs(args) print(f"scenario : {scenario} (excel={do_excel}, report={'v4' if do_report else 'none'})") # Surface data completeness so the Agent never presents a partial report as complete. if _DATA_QUALITY.incomplete: print(f"DATA_INCOMPLETE: {_DATA_QUALITY.summary()}. " f"Some time-windows/batches failed after retries; figures are UNDER-counted. " f"Recommend re-running the report.", flush=True) else: print("data_quality : complete (all windows fetched)", flush=True) if do_excel: excel_path = out_dir / ( f"miravia_orders_{fname_range}_{ts}." + ('csv' if args.format == 'csv' else 'xlsx') ) excel_path = _resolve_collision(excel_path) export_dataframe(rows, excel_path, args.format) print(f"output : {excel_path}") if do_report: # Phase 1: Compute report data payload and metrics (NO HTML yet). # HTML rendering is deferred to render_final.py AFTER Agent generates # and validates insights.json. This ensures the final HTML always # contains a complete AI Executive Summary. import report_v4 # Compute pre-aggregated report data (window.DATA payload) report_data = report_v4.prepare_report_data( rows=rows, range_label=range_label, export_columns=EXPORT_COLUMNS, lang=getattr(args, 'lang', 'en'), data_quality=_DATA_QUALITY.to_dict(), ) report_data_path = out_dir / f"report_data_{fname_range}_{ts}.json" report_data_path = _resolve_collision(report_data_path) report_data_path.write_text( json.dumps(report_data, ensure_ascii=False, indent=2, default=str), encoding="utf-8", ) print(f"report_data : {report_data_path}") # Export structured metrics for Agent LLM analysis metrics_data = report_v4.export_metrics( rows=rows, export_columns=EXPORT_COLUMNS, range_start=str(after), range_end=str(before), ) metrics_path = out_dir / f"metrics_{fname_range}_{ts}.json" metrics_path = _resolve_collision(metrics_path) metrics_path.write_text( json.dumps(metrics_data, ensure_ascii=False, indent=2, default=str), encoding="utf-8", ) print(f"metrics : {metrics_path}") # ★ Hard gate: Agent MUST generate insights and call render_final.py # Generate unique insights path with same timestamp to avoid stale file contamination insights_path = out_dir / f"insights_{fname_range}_{ts}.json" insights_path = _resolve_collision(insights_path) print(f"insights_target : {insights_path}") print(f"NEXT_ACTION_REQUIRED: python3 scripts/render_final.py --report-data {report_data_path} --insights {insights_path} --output {out_dir} --lang {getattr(args, 'lang', 'en')}") print(f"TASK_INCOMPLETE: Data exported. Agent must now generate insights to {insights_path} from metrics, then call render_final.py to produce the final HTML report.") # Write skill invocation log _write_run_log( start_ms=_start_ms, scenario=scenario, data_count=total_sub_orders, remark=f"range={after}~{before} hit_cap={hit_cap}", ) return 2 if hit_cap else 0 def _write_run_log(start_ms: int, scenario: str, data_count: int, error_code: str = "", remark: str = "") -> None: """Write skill log after run() completes normally.""" _duration = int(time.time() * 1000) - start_ms status = "success" if not error_code else "fail" write_skill_log( action=scenario, status=status, duration_ms=_duration, data_count=data_count, error_code=error_code, remark=remark, ) def _open_in_browser(html_path: Path) -> None: """Open HTML report in user's default browser (cross-platform with fallback). Prefers webbrowser module; falls back to platform commands (macOS: `open`, Linux: `xdg-open`, Windows: `os.startfile`). Any exception only prints a warning without affecting main exit code. """ import webbrowser import subprocess import platform url = html_path.resolve().as_uri() # file:///... try: if webbrowser.open(url, new=2): print(f"opened : {url}") return except Exception as e: # pragma: no cover - browser env-specific print(f"warning: webbrowser.open failed: {e}", file=sys.stderr) # Fallback: invoke platform command try: sysname = platform.system() if sysname == "Darwin": subprocess.Popen(["open", str(html_path)]) elif sysname == "Windows": # pragma: no cover os.startfile(str(html_path)) # type: ignore[attr-defined] else: subprocess.Popen(["xdg-open", str(html_path)]) print(f"opened : {url}") except Exception as e: # pragma: no cover print(f"warning: failed to open browser ({e}); please open manually: {html_path}", file=sys.stderr) def _resolve_outputs(args: argparse.Namespace): """Determine outputs based on --scenario / --with-excel / --report. Returns (scenario, do_excel, do_report). - scenario: 'export' | 'refund' | 'business' | 'auto' - do_excel / do_report: whether to produce the corresponding output v4 unified dashboard: all scenarios use the same HTML template (report_v4.html), no longer distinguish refund / business report types. """ scenario = (args.scenario or "auto").lower() report_flag = str(args.report).lower() not in ("false", "0", "no") if scenario == "export": # Scenario 1: Excel/CSV only return scenario, True, False if scenario == "refund": # Scenario 2: Refund-focused HTML (v4 unified dashboard); Excel only with --with-excel return scenario, bool(args.with_excel), report_flag if scenario == "business": # Scenario 3: Business analysis HTML (v4 unified dashboard); Excel only with --with-excel return scenario, bool(args.with_excel), report_flag # auto: backward-compatible behavior -- Excel and v4 unified dashboard HTML both output return "auto", True, report_flag # Valid status values supported by /orders/get API (test-case-required subset). # See: .qoder/doc/tech/order_api_reference.md for full API reference. VALID_STATUSES = [ "all", # all orders (no filter) # -- aggregated statuses -- "all_to_ship", # To Ship: topack + packed + rts "all_shipped", # Shipped: shipped + delivered # -- specific statuses -- "topack", # to pack (待打包) "packed", # packed (已打包) "rts", # ready to ship / processing (待揽收) "canceled", # canceled (已取消) "shipped", # shipped / picked up (已发货/已揽收) "delivered", # delivered (已妥投) ] # CLI-friendly aliases (currently none) STATUS_ALIASES = {} def _resolve_api_status(cli_status: str) -> Optional[str]: """Map CLI --status value to the actual API parameter. Returns None for 'all' (meaning no status filter). Resolves aliases (e.g. 'to_ship' → 'ready_to_ship'). Other values are passed through directly to the API. """ if cli_status == "all": return None return STATUS_ALIASES.get(cli_status, cli_status) def _build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser( prog="miravia-order-report", description="Export Miravia seller orders and generate analysis report.", ) p.add_argument("--created-after") p.add_argument("--created-before") p.add_argument("--last-days", type=int, default=None, help="convenience: include today and previous N-1 days (whole-day aligned). Overrides --created-after/before.") p.add_argument("--status", default="all", choices=VALID_STATUSES, help="Order status filter (default: all). " "Aggregated: all_to_ship, all_shipped. " "Specific: topack, packed, rts, to_respond, canceled, shipped, delivered, failed_delivery, lost") p.add_argument("--channel", default=None, choices=["AE", "Miravia"], help="Channel filter: AE (M2A/AliExpress) or Miravia") p.add_argument("--ship-to", default=None, help="Destination country filter (e.g. ES, PT, IT)") p.add_argument("--format", default="xlsx", choices=["xlsx", "csv"]) p.add_argument("--output", default="./out") p.add_argument("--max-orders", type=int, default=config.DEFAULT_MAX_ORDERS, help="Max sub-orders per export (default 10000). Exports partial data and prompts user to narrow range if exceeded.") p.add_argument("--scenario", default="auto", choices=["auto", "export", "refund", "business", "inspect"], help="Output scenario: export=Excel only; refund=refund-focused HTML; " "business=business analysis HTML; auto=Excel+unified HTML (backward-compat); " "inspect=single-order diagnostic text (stdout, no file output)") p.add_argument("--order-ids", default=None, help="Comma-separated trade order IDs (e.g. '12345,67890'). " "Works with all scenarios: inspect=stdout text, others=fast-path Excel/HTML. " "When provided, time range is not required.") p.add_argument("--with-excel", action="store_true", help="Additionally output Excel details in refund/business scenarios") p.add_argument("--report", default="true", help="HTML report master switch (true/false); only effective when scenario != export") p.add_argument("--ping", action="store_true", help="health check only, no data fetch") # By default, auto-open HTML report in browser after generation; --no-open to disable p.add_argument("--open", dest="open_browser", action="store_true", default=True, help="Auto-open HTML report in default browser after generation (enabled by default)") p.add_argument("--no-open", dest="open_browser", action="store_false", help="Disable auto-opening HTML report") p.add_argument("--lang", default="en", help="HTML report language code (default: en). Falls back to English if unsupported.") p.add_argument("--agent-type", required=True, help="Agent/IDE type identifier (required), dynamically passed by the calling Agent, e.g. qoder / cursor / claude / custom_bot") return p def main() -> int: logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") args = _build_parser().parse_args() # Set global agent_type (CLI required, sole source) set_agent_type(args.agent_type) # Keep args.report string raw value for _resolve_outputs; also provide boolean alias args.report_enabled = str(args.report).lower() not in ("false", "0", "no") try: return run(args) except ValueError as e: print(f"error: invalid argument: {e}", file=sys.stderr) print( "hint: --created-after/--created-before expect 'YYYY-MM-DD' or full ISO 8601.", file=sys.stderr, ) write_skill_log(action=(args.scenario or "auto"), status="fail", duration_ms=0, error_code="INVALID_ARGUMENT", remark=str(e)) return 1 except MiraviaError as e: print(f"error: {e}", file=sys.stderr) write_skill_log(action=(args.scenario or "auto"), status="fail", duration_ms=0, error_code=e.code, remark=str(e)) return 1 if __name__ == "__main__": sys.exit(main())