Linkfox Ruiguan Patent Design

#!/usr/bin/env python3
"""
Skill response I/O helper — wraps any main script to persist large API
responses to disk, then offers a `read` subcommand to extract specific fields
from those persisted files. Generic, business-agnostic.

This script is bundled into each skill's scripts/ directory by tools/response_io/sync.py.
The agent must pass --script <path> to identify which main script to execute.

Usage:
  python scripts/response_io.py run --script <PATH> --out-dir <DIR> '<json_params>' [--label NAME] [--timeout SEC]
  python scripts/response_io.py read <file> (--path "<JMESPath>" | --fields "f1,f2,...") [--limit N] [--offset M] [--format json|jsonl|csv|table]
"""

from __future__ import annotations

import sys
if sys.version_info < (3, 10):
    sys.exit(
        "Error: Python 3.10+ required (current: "
        f"{sys.version_info.major}.{sys.version_info.minor}). "
        "Please upgrade Python."
    )

import argparse
import csv
import io
import json
import os
import re
import secrets
import subprocess
from datetime import datetime
from pathlib import Path
from typing import Any

# Force UTF-8 stdout/stderr so non-ASCII chars in previews and API responses
# print correctly on Windows (default cp936 / gbk).
for stream in (sys.stdout, sys.stderr):
    try:
        stream.reconfigure(encoding="utf-8")  # type: ignore[attr-defined]
    except (AttributeError, OSError):
        pass

try:
    import jmespath  # type: ignore
    HAS_JMESPATH = True
except ImportError:
    HAS_JMESPATH = False


MAX_STRING_LEN = 120
MAX_DEPTH = 3
SAMPLE_KEY_CAP = 15
RAW_TEXT_PEEK = 500
DEFAULT_TIMEOUT_SEC = 300


# ---------------------------------------------------------------------------
# Shared helpers
# ---------------------------------------------------------------------------


def _err(msg: str, code: int = 1) -> None:
    print(msg, file=sys.stderr)
    sys.exit(code)


def _resolve_script(script_arg: str) -> Path:
    p = Path(script_arg).expanduser()
    if not p.is_absolute():
        # Resolve relative to the current working directory the agent invoked from.
        p = (Path.cwd() / p).resolve()
    else:
        p = p.resolve()
    if not p.is_file():
        _err(f"--script path not found: {p}")
    return p


def _resolve_skill_name(main_script: Path) -> str:
    """Best-effort skill name extraction for filename prefixing.

    main_script lives at <skill_dir>/scripts/<name>.py — return <skill_dir>'s
    folder name. Fall back to the script's stem if structure differs.
    """
    try:
        if main_script.parent.name == "scripts":
            return main_script.parents[1].name
    except IndexError:
        pass
    return main_script.stem


def _sanitize_label(label: str) -> str:
    """Allow only safe filename chars in --label to prevent path traversal."""
    cleaned = re.sub(r"[^\w\-]", "_", label)
    return cleaned[:64]  # cap length


def _truncate_string(s: str) -> str:
    if len(s) <= MAX_STRING_LEN:
        return s
    return s[:MAX_STRING_LEN] + f"...(truncated, total {len(s)} chars)"


def _truncate_value(value: Any, depth: int = 0) -> Any:
    """Recursively truncate strings, deep nesting, and large arrays for preview."""
    if depth >= MAX_DEPTH:
        if isinstance(value, dict):
            return f"<truncated nested object, keys: {list(value.keys())[:10]}>"
        if isinstance(value, list):
            return f"<truncated nested array, length: {len(value)}>"
        if isinstance(value, str):
            return _truncate_string(value)
        return value
    if isinstance(value, str):
        return _truncate_string(value)
    if isinstance(value, dict):
        out = {k: _truncate_value(v, depth + 1) for k, v in value.items()}
        return out
    if isinstance(value, list):
        if not value:
            return []
        truncated = [_truncate_value(value[0], depth + 1)]
        if len(value) > 1:
            # Note total length on the parent — keep the array type-homogeneous
            # so downstream consumers can iterate without special-casing strings.
            truncated.append({"_omitted_items": len(value) - 1})
        return truncated
    return value


def _shape_of(value: Any, top: bool = False) -> Any:
    """Lightweight schema description for the preview block."""
    if isinstance(value, dict):
        keys = list(value.keys())
        out: dict[str, Any] = {"type": "object", "top_keys" if top else "keys": keys}
        if top:
            for k in keys[:8]:
                out[k] = _shape_of(value[k])
        return out
    if isinstance(value, list):
        out = {"type": "array", "length": len(value)}
        if value and isinstance(value[0], dict):
            out["item_keys"] = list(value[0].keys())
        elif value:
            out["item_type"] = type(value[0]).__name__
        return out
    return {"type": type(value).__name__}


def _build_sample(value: Any) -> Any:
    """First-record sample with explicit truncation marker."""
    if isinstance(value, list):
        if not value:
            return {"_truncated_record": True, "_note": "array is empty"}
        first = value[0]
        if isinstance(first, dict):
            sample = {"_truncated_record": True, "_note": f"first of {len(value)} items"}
            sample.update(_truncate_value(first, depth=1))
            return sample
        return {"_truncated_record": True, "_note": f"first of {len(value)} items", "value": _truncate_value(first, depth=1)}
    if isinstance(value, dict):
        sample = {"_truncated_record": True, "_note": "top-level object (truncated)"}
        sample.update(_truncate_value(value, depth=1))
        return sample
    return {"_truncated_record": True, "value": _truncate_value(value, depth=1)}


def _shrink_preview(preview: dict) -> dict:
    """Cap the sample's value fields when it has many keys.

    `shape.*.item_keys` is the single source of truth for the full key list
    (always complete, no truncation). The sample only ever shows up to
    SAMPLE_KEY_CAP fields with their concrete values, since the agent only
    needs a feel for value shapes — for the full menu of available fields,
    they read `shape`.
    """
    sample = preview.get("sample")
    if isinstance(sample, dict):
        meta_keys = {"_truncated_record", "_note"}
        data_keys = [k for k in sample.keys() if k not in meta_keys]
        if len(data_keys) > SAMPLE_KEY_CAP:
            kept = data_keys[:SAMPLE_KEY_CAP]
            new_sample = {k: v for k, v in sample.items() if k in meta_keys or k in kept}
            base_note = sample.get("_note", "")
            extra = (
                f"showing first {SAMPLE_KEY_CAP} of {len(data_keys)} fields "
                f"(see `shape` for the complete key list)"
            )
            new_sample["_note"] = f"{base_note}; {extra}" if base_note else extra
            preview["sample"] = new_sample
    return preview


# ---------------------------------------------------------------------------
# `run` subcommand
# ---------------------------------------------------------------------------


def cmd_run(args: argparse.Namespace) -> int:
    main_script = _resolve_script(args.script)
    skill_name = _resolve_skill_name(main_script)

    out_dir = Path(args.out_dir).expanduser().resolve()
    try:
        out_dir.mkdir(parents=True, exist_ok=True)
    except OSError as e:
        _err(f"Failed to create --out-dir {out_dir}: {e}")
    if not os.access(out_dir, os.W_OK):
        _err(f"--out-dir is not writable: {out_dir}")

    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    rand = secrets.token_hex(3)
    safe_label = _sanitize_label(args.label) if args.label else ""
    label_part = f"__{safe_label}" if safe_label else ""
    out_file = out_dir / f"{skill_name}__{timestamp}_{rand}{label_part}.json"

    # Force the child process to emit UTF-8 regardless of the host console
    # encoding (Windows defaults to cp936 / gbk and would otherwise corrupt
    # non-ASCII bytes when we read them back).
    child_env = os.environ.copy()
    child_env["PYTHONIOENCODING"] = "utf-8"

    timed_out = False
    try:
        proc = subprocess.run(
            [sys.executable, str(main_script), args.params],
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
            env=child_env,
            timeout=args.timeout,
        )
        stdout_text = proc.stdout or ""
        stderr_text = proc.stderr or ""
        returncode = proc.returncode
    except subprocess.TimeoutExpired as e:
        timed_out = True
        stdout_text = (e.stdout.decode("utf-8", errors="replace") if isinstance(e.stdout, bytes) else (e.stdout or "")) or ""
        stderr_text = (e.stderr.decode("utf-8", errors="replace") if isinstance(e.stderr, bytes) else (e.stderr or "")) or ""
        returncode = 124  # convention for timeout

    # Always write the captured stdout to disk, even if not JSON.
    try:
        out_file.write_text(stdout_text, encoding="utf-8")
    except OSError as e:
        _err(f"Failed to write output file {out_file}: {e}")

    if stderr_text:
        sys.stderr.write(stderr_text)

    # Try to parse the captured stdout as JSON for the preview.
    try:
        parsed = json.loads(stdout_text) if stdout_text.strip() else None
        format_kind = "json"
    except json.JSONDecodeError:
        parsed = None
        format_kind = "raw_text"

    preview: dict[str, Any] = {
        "_preview": {
            "is_preview": True,
            "warning": (
                "PREVIEW ONLY — NOT FULL DATA. The full response is saved to `file`. "
                "Use `python scripts/response_io.py read <file> --fields '...'` to extract "
                "specific fields, or `--path '<JMESPath>'` for complex projections."
            ),
        },
    }

    # Surface failures prominently so agents don't mistake a stub preview for success.
    if returncode != 0 or timed_out:
        stderr_snippet = stderr_text[-500:] if stderr_text else ""
        preview["_error"] = {
            "exit_code": returncode,
            "timed_out": timed_out,
            "stderr_snippet": stderr_snippet,
            "hint": "The wrapped script failed or timed out. The output file may be empty or partial.",
        }

    preview.update({
        "file": str(out_file),
        "size_bytes": out_file.stat().st_size,
        "skill": skill_name,
        "exit_code": returncode,
        "format": format_kind,
        "label": safe_label or None,
        "next_steps_hint": (
            "use: python scripts/response_io.py read <file> --fields '...' | --path '...'"
        ),
    })

    if format_kind == "json":
        preview["shape"] = _shape_of(parsed, top=True)
        preview["sample"] = _build_sample(parsed)
    else:
        peek = stdout_text[:RAW_TEXT_PEEK]
        preview["raw_text_peek"] = peek
        preview["raw_text_total_chars"] = len(stdout_text)
        preview["sample"] = {
            "_truncated_record": True,
            "_note": f"stdout was not valid JSON; first {RAW_TEXT_PEEK} chars shown above in raw_text_peek",
        }

    preview = _shrink_preview(preview)
    print(json.dumps(preview, ensure_ascii=False, indent=2))
    return returncode


# ---------------------------------------------------------------------------
# `read` subcommand
# ---------------------------------------------------------------------------


def _load_json(path: Path) -> Any:
    try:
        text = path.read_text(encoding="utf-8")
    except OSError as e:
        _err(f"Failed to read file {path}: {e}")
    try:
        return json.loads(text)
    except json.JSONDecodeError as e:
        _err(f"File is not valid JSON: {path}\n{e}")


def _basic_dot_path(data: Any, path: str) -> Any:
    """Pure-stdlib dot-path resolver. No [*] support — callers fall back here only when jmespath is unavailable AND the path has no [*]."""
    cur = data
    for part in path.split("."):
        if isinstance(cur, dict):
            cur = cur.get(part)
        else:
            return None
    return cur


def _resolve_field(data: Any, expr: str) -> Any:
    if HAS_JMESPATH:
        return jmespath.search(expr, data)
    if "[" in expr or "*" in expr:
        _err(
            f"jmespath is required for expression '{expr}'. "
            f"Install with: pip install jmespath"
        )
    return _basic_dot_path(data, expr)


def _project_fields(data: Any, fields: list[str]) -> Any:
    """Run each field expr; if any returns a list, zip them into list-of-dicts."""
    resolved: dict[str, Any] = {f: _resolve_field(data, f) for f in fields}

    list_lengths = [len(v) for v in resolved.values() if isinstance(v, list)]
    if not list_lengths:
        return resolved

    # All list values must be same length to zip cleanly.
    if len(set(list_lengths)) > 1:
        # Fallback: return the dict as-is so caller can inspect mismatches.
        return resolved

    n = list_lengths[0]
    rows = []
    for i in range(n):
        row = {}
        for f, v in resolved.items():
            row[f] = v[i] if isinstance(v, list) else v
        rows.append(row)
    return rows


def _apply_slice(value: Any, limit: int | None, offset: int | None) -> Any:
    if not isinstance(value, list):
        return value
    start = offset or 0
    end = (start + limit) if limit is not None else None
    return value[start:end]


def _format_output(value: Any, fmt: str) -> str:
    if fmt == "json":
        return json.dumps(value, ensure_ascii=False, indent=2)
    if fmt == "jsonl":
        if isinstance(value, list):
            return "\n".join(json.dumps(item, ensure_ascii=False) for item in value)
        return json.dumps(value, ensure_ascii=False)
    if fmt in ("csv", "table"):
        if not isinstance(value, list) or not value:
            _err(f"--format {fmt} requires a non-empty list result")
        if not all(isinstance(item, dict) for item in value):
            _err(f"--format {fmt} requires list-of-objects, got list of {type(value[0]).__name__}")
        keys: list[str] = []
        for item in value:
            for k in item.keys():
                if k not in keys:
                    keys.append(k)
        if fmt == "csv":
            buf = io.StringIO()
            writer = csv.DictWriter(buf, fieldnames=keys, extrasaction="ignore")
            writer.writeheader()
            for item in value:
                writer.writerow({k: _stringify(item.get(k)) for k in keys})
            return buf.getvalue().rstrip("\n")
        # table: simple aligned columns
        rows = [[_stringify(item.get(k)) for k in keys] for item in value]
        widths = [len(k) for k in keys]
        for row in rows:
            for i, cell in enumerate(row):
                widths[i] = max(widths[i], len(cell))
        lines = [
            "  ".join(k.ljust(widths[i]) for i, k in enumerate(keys)),
            "  ".join("-" * widths[i] for i in range(len(keys))),
        ]
        for row in rows:
            lines.append("  ".join(row[i].ljust(widths[i]) for i in range(len(keys))))
        return "\n".join(lines)
    _err(f"Unknown --format: {fmt}")
    return ""  # unreachable


def _stringify(v: Any) -> str:
    if v is None:
        return ""
    if isinstance(v, (dict, list)):
        return json.dumps(v, ensure_ascii=False)
    return str(v)


def cmd_read(args: argparse.Namespace) -> int:
    if not args.path and not args.fields:
        _err("read: either --path or --fields is required")
    if args.path and args.fields:
        _err("read: --path and --fields are mutually exclusive")

    file_path = Path(args.file).expanduser().resolve()
    data = _load_json(file_path)

    if args.path:
        result = _resolve_field(data, args.path)
    else:
        fields = [f.strip() for f in args.fields.split(",") if f.strip()]
        if not fields:
            _err("--fields parsed to empty list")
        result = _project_fields(data, fields)

    result = _apply_slice(result, args.limit, args.offset)
    print(_format_output(result, args.format))
    return 0


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------


def main() -> int:
    parser = argparse.ArgumentParser(
        prog="response_io.py",
        description="Persist large skill API responses to disk and read fields on demand.",
    )
    sub = parser.add_subparsers(dest="cmd", required=True)

    p_run = sub.add_parser(
        "run",
        help="Execute a main script and persist its stdout to a file; "
             "print only a lightweight preview to stdout.",
    )
    p_run.add_argument("params", help="JSON params string passed verbatim to the main script (argv[1]).")
    p_run.add_argument("--script", required=True, help="Path to the main script to execute, e.g. scripts/my_api.py")
    p_run.add_argument("--out-dir", required=True, help="Directory to write the response file into (created if missing).")
    p_run.add_argument("--label", default=None, help="Optional filename suffix; sanitized to safe filename characters.")
    p_run.add_argument("--timeout", type=int, default=DEFAULT_TIMEOUT_SEC, help=f"Subprocess timeout in seconds (default: {DEFAULT_TIMEOUT_SEC}).")
    p_run.set_defaults(func=cmd_run)

    p_read = sub.add_parser(
        "read",
        help="Extract specific fields from a previously persisted response file.",
    )
    p_read.add_argument("file", help="Path to the persisted JSON response file.")
    g = p_read.add_mutually_exclusive_group()
    g.add_argument("--path", default=None, help="JMESPath expression, e.g. 'data[*].{asin: asin, title: title}'.")
    g.add_argument("--fields", default=None, help="Comma-separated field paths, e.g. 'data[*].asin,data[*].title'.")
    p_read.add_argument("--limit", type=int, default=None, help="Take at most N items (when result is a list).")
    p_read.add_argument("--offset", type=int, default=None, help="Skip the first M items (when result is a list).")
    p_read.add_argument("--format", choices=["json", "jsonl", "csv", "table"], default="json", help="Output format (default: json).")
    p_read.set_defaults(func=cmd_read)

    args = parser.parse_args()
    return args.func(args)


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""
Ruiguan Design Patent Detection - LinkFox Skill
Calls the ruiguan/detectionPatentDesign API endpoint to check product images
against global design patent databases.

Usage:
  python ruiguan_detection_patent_design.py '{"imageUrl": "https://example.com/product.jpg", "queryMode": "hybrid", "topNumber": 50, "regions": "US"}'
"""

import json
import os
import sys
from urllib.request import urlopen, Request
from urllib.error import HTTPError, URLError


API_URL = "https://tool-gateway.linkfox.com/ruiguan/detectionPatentDesign"

# Required parameters that must be present in the request
REQUIRED_PARAMS = ["imageUrl", "queryMode", "topNumber"]

# Default values applied when optional parameters are not provided
DEFAULTS = {
    "queryMode": "hybrid",
    "topNumber": 100,
    "regions": "US",
    "patentStatus": "1",
    "enableRadar": True,
}


def get_api_key():
    """Retrieve the API key from environment, with a friendly prompt if missing."""
    key = os.environ.get("LINKFOXAGENT_API_KEY")
    if not key:
        print(
            "API Key not configured. Please complete authorization first:\n"
            "1. Visit https://skill.linkfox.com/linkfoxskills/guide.htm to obtain your Key\n"
            "2. Set the environment variable: export LINKFOXAGENT_API_KEY=your-key-here",
            file=sys.stderr,
        )
        sys.exit(1)
    return key


def validate_params(params: dict):
    """Validate that all required parameters are present."""
    missing = [p for p in REQUIRED_PARAMS if p not in params]
    if missing:
        print(
            f"Missing required parameters: {', '.join(missing)}\n"
            f"Required: imageUrl (product image URL), queryMode (physical/line/hybrid), topNumber (1-100)",
            file=sys.stderr,
        )
        sys.exit(1)

    # Validate topNumber range
    top_number = params.get("topNumber", 100)
    if not isinstance(top_number, int) or top_number < 1 or top_number > 100:
        print("topNumber must be an integer between 1 and 100", file=sys.stderr)
        sys.exit(1)

    # Validate queryMode value
    query_mode = params.get("queryMode", "hybrid")
    if query_mode not in ("physical", "line", "hybrid"):
        print(
            f"Invalid queryMode: '{query_mode}'. Must be one of: physical, line, hybrid",
            file=sys.stderr,
        )
        sys.exit(1)


def apply_defaults(params: dict) -> dict:
    """Apply default values for optional parameters that are not provided."""
    result = dict(params)
    for key, default_value in DEFAULTS.items():
        if key not in result:
            result[key] = default_value
    return result


def call_api(params: dict) -> dict:
    """Call the Ruiguan design patent detection API."""
    api_key = get_api_key()
    data = json.dumps(params).encode("utf-8")

    req = Request(
        API_URL,
        data=data,
        headers={
            "Authorization": api_key,
            "Content-Type": "application/json",
            "User-Agent": "LinkFox-Skill/1.0",
        },
        method="POST",
    )

    try:
        # Use a longer timeout since patent detection can be slow
        with urlopen(req, timeout=120) as response:
            return json.loads(response.read().decode("utf-8"))
    except HTTPError as e:
        body = e.read().decode("utf-8") if e.fp else ""
        return {"error": f"HTTP {e.code}: {e.reason}", "details": body}
    except URLError as e:
        return {"error": f"Connection failed: {e.reason}"}


def summarize_results(result: dict):
    """Print a human-readable summary of high-risk patents."""
    data = result.get("data", [])
    total = result.get("total", 0)

    print(f"\n--- Detection Summary ---")
    print(f"Total patents found: {total}")

    # Filter high-risk patents (similarity >= 0.7 or has TRO history)
    high_risk = []
    for patent in data:
        similarity = float(patent.get("similarity", 0))
        tro_case = patent.get("troCase", False)
        tro_holder = patent.get("troHolder", False)
        if similarity >= 0.7 or tro_case or tro_holder:
            high_risk.append(patent)

    if high_risk:
        print(f"High-risk patents (similarity >= 0.7 or TRO history): {len(high_risk)}")
        print()
        for i, patent in enumerate(high_risk, 1):
            print(f"  [{i}] {patent.get('applicationNumber', 'N/A')}")
            print(f"      Title: {patent.get('patentProd', 'N/A')}")
            print(f"      Title (CN): {patent.get('patentProdCn', 'N/A')}")
            print(f"      Similarity: {patent.get('similarity', 'N/A')}")
            print(f"      TRO Case: {patent.get('troCase', False)}")
            print(f"      TRO Holder: {patent.get('troHolder', False)}")
            radar = patent.get("radarResult", {})
            if radar:
                print(f"      Radar - Suspected Infringement: {radar.get('same', 'N/A')}")
                print(f"      Radar - Explanation: {radar.get('exp', 'N/A')}")
            print()
    else:
        print("No high-risk patents found (none with similarity >= 0.7 or TRO history).")

    print("Note: This detection result is generated by LinkfoxAgent.")
    print("It is recommended to consult a professional IP attorney for legal advice.")


def main():
    if len(sys.argv) < 2:
        print("Usage: ruiguan_detection_patent_design.py '<JSON parameters>'", file=sys.stderr)
        print(
            "Example: ruiguan_detection_patent_design.py "
            "'{\"imageUrl\": \"https://example.com/product.jpg\", \"queryMode\": \"hybrid\", \"topNumber\": 50}'",
            file=sys.stderr,
        )
        print(
            "\nRequired parameters:\n"
            "  imageUrl   - Product image URL\n"
            "  queryMode  - Search mode: physical, line, or hybrid\n"
            "  topNumber  - Number of results (1-100)\n"
            "\nOptional parameters:\n"
            "  regions           - Country codes, e.g., US,EU,CN (default: US)\n"
            "  productTitle      - Product title\n"
            "  productDescription - Product description\n"
            "  patentStatus      - 1 (active), 0 (expired), 1,0 (both) (default: 1)\n"
            "  enableRadar       - Enable radar analysis (default: true)\n"
            "  topLoc            - LOC codes, e.g., 06,07\n"
            "  sourceLanguage    - Source language, e.g., zh-CN",
            file=sys.stderr,
        )
        sys.exit(1)

    try:
        params = json.loads(sys.argv[1])
    except json.JSONDecodeError as e:
        print(f"Invalid parameter format: {e}", file=sys.stderr)
        sys.exit(1)

    # Validate required parameters
    validate_params(params)

    # Apply defaults for optional parameters
    params = apply_defaults(params)

    # Call the API
    result = call_api(params)

    # Print full JSON response
    print(json.dumps(result, indent=2, ensure_ascii=False))

    # If successful and contains data, print a summary
    if "error" not in result and "data" in result:
        summarize_results(result)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Upload Local Image - LinkFox Skill
Uploads a local image file to LinkFox OSS and returns a publicly accessible URL.

Steps:
  1. Request a presigned PUT URL from the LinkFox OSS gateway
  2. Upload the local file to the presigned URL
  3. Return the public URL (valid for 24 hours)

Usage:
  python upload_image.py /path/to/local/image.png
"""

import json
import os
import sys
from urllib.request import urlopen, Request
from urllib.error import HTTPError, URLError

PRESIGN_URL = "https://tool-gateway.linkfox.com/oss/file/presignedPut"

CONTENT_TYPE_MAP = {
    "jpg": "image/jpeg",
    "jpeg": "image/jpeg",
    "png": "image/png",
    "gif": "image/gif",
    "webp": "image/webp",
    "heic": "image/heic",
}


def get_api_key():
    """Retrieve the API key from environment, with a friendly prompt if missing."""
    key = os.environ.get("LINKFOXAGENT_API_KEY")
    if not key:
        print(
            "API Key not configured. Please complete authorization first:\n"
            "1. Visit https://skill.linkfox.com/linkfoxskills/guide.htm to obtain your Key\n"
            "2. Set the environment variable: export LINKFOXAGENT_API_KEY=your-key-here",
            file=sys.stderr,
        )
        sys.exit(1)
    return key


def get_presigned_url(content_type: str, file_extension: str) -> str:
    """Request a presigned PUT URL from the OSS gateway."""
    api_key = get_api_key()
    data = json.dumps({
        "contentType": content_type,
        "fileExtension": file_extension,
    }).encode("utf-8")

    req = Request(
        PRESIGN_URL,
        data=data,
        headers={
            "Authorization": api_key,
            "Content-Type": "application/json",
            "User-Agent": "LinkFox-Skill/1.0",
        },
        method="POST",
    )

    try:
        with urlopen(req, timeout=30) as response:
            result = json.loads(response.read().decode("utf-8"))
    except HTTPError as e:
        body = e.read().decode("utf-8") if e.fp else ""
        print(f"Failed to get presigned URL: HTTP {e.code}: {e.reason}\n{body}", file=sys.stderr)
        sys.exit(1)
    except URLError as e:
        print(f"Connection failed: {e.reason}", file=sys.stderr)
        sys.exit(1)

    if result.get("errcode") != 200:
        print(f"API error: {result.get('errmsg', 'unknown error')}", file=sys.stderr)
        sys.exit(1)

    return result["url"]


def upload_file(presigned_url: str, file_path: str, content_type: str):
    """Upload the local file to the presigned OSS URL via HTTP PUT."""
    with open(file_path, "rb") as f:
        file_data = f.read()

    req = Request(
        presigned_url,
        data=file_data,
        headers={
            "Content-Type": content_type,
            "x-oss-object-acl": "public-read",
        },
        method="PUT",
    )

    try:
        with urlopen(req, timeout=120) as response:
            if response.status not in (200, 201):
                print(f"Upload failed with status: {response.status}", file=sys.stderr)
                sys.exit(1)
    except HTTPError as e:
        body = e.read().decode("utf-8") if e.fp else ""
        print(f"Upload failed: HTTP {e.code}: {e.reason}\n{body}", file=sys.stderr)
        sys.exit(1)
    except URLError as e:
        print(f"Upload connection failed: {e.reason}", file=sys.stderr)
        sys.exit(1)


def extract_public_url(presigned_url: str) -> str:
    """Extract the base public URL by stripping query parameters."""
    return presigned_url.split("?")[0]


def main():
    if len(sys.argv) < 2:
        print(
            "Usage: upload_image.py <local_image_path>\n"
            "Example: upload_image.py /path/to/product.png",
            file=sys.stderr,
        )
        sys.exit(1)

    file_path = sys.argv[1]

    if not os.path.isfile(file_path):
        print(f"File not found: {file_path}", file=sys.stderr)
        sys.exit(1)

    extension = os.path.splitext(file_path)[1].lstrip(".").lower()
    content_type = CONTENT_TYPE_MAP.get(extension)
    if not content_type:
        print(
            f"Unsupported image format: .{extension}\n"
            f"Supported formats: {', '.join(CONTENT_TYPE_MAP.keys())}",
            file=sys.stderr,
        )
        sys.exit(1)

    # Step 1: Get presigned URL
    presigned_url = get_presigned_url(content_type, extension)

    # Step 2: Upload file
    upload_file(presigned_url, file_path, content_type)

    # Step 3: Output public URL
    public_url = extract_public_url(presigned_url)
    print(json.dumps({"url": public_url}, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    main()