واجهة برمجة التطبيقات العامة (API)

"""
Bas-Relief Public API (Python) — production sample

Recommended: provide your API key via environment variable.

Windows (PowerShell):
  $env:BASRELIEF_API_KEY = "YOUR_API_KEY"

macOS/Linux (bash/zsh):
  export BASRELIEF_API_KEY="YOUR_API_KEY"
"""

import base64
import json
import os
import sys
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, List, Optional
from urllib.error import HTTPError, URLError
from urllib.parse import urljoin, urlparse
from urllib.request import Request, urlopen

DEFAULT_BASE_URL = "https://bas-relief.ai"
BASE_URL: str = DEFAULT_BASE_URL
APPROACH: str = "A"

# =====================
# EDIT THESE SETTINGS
# =====================

API_KEY: str = os.getenv("BASRELIEF_API_KEY", "").strip()

# Job inputs
SUBJECT: str = "your subject here"
TEXT_TOP: str = ""
TEXT_CENTER: str = "your text here"
TEXT_BOTTOM: str = ""
TEXT_BOTTOM2: str = ""

LOCALE: str = "en"
METHOD_NO: str = "C"

# Style: choose ONE of these options
STYLE_NO: int = 41 
CUSTOM_STYLE_TEXT: Optional[str] = None 

# Size (optional)
WIDTH: Optional[int] = 1024
HEIGHT: Optional[int] = 1024

MASK_PATH: Optional[str] = None
CLIENT_REFERENCE: Optional[str] = None
OUTDIR: str = str(Path.cwd() / "downloads")

TIMEOUT_SECONDS: int = 15 * 60
POLL_SECONDS: float = 2.0

@dataclass(frozen=True)
class ApiConfig:
    base_url: str
    api_key: str
    timeout_seconds: int
    poll_seconds: float

class ApiError(RuntimeError):
    pass

def _json_request(method: str, url: str, *, headers: Dict[str, str], body: Optional[Dict[str, Any]] = None, timeout: int = 60) -> Any:
    data = None
    req_headers = {"Accept": "application/json", **headers}
    if body is not None:
        data = json.dumps(body).encode("utf-8")
        req_headers["Content-Type"] = "application/json"
    
    req = Request(url, method=method, headers=req_headers, data=data)
    try:
        with urlopen(req, timeout=timeout) as resp:
            raw = resp.read()
            return json.loads(raw.decode("utf-8")) if raw else None
    except HTTPError as e:
        raw = e.read() if e else None
        detail = json.loads(raw.decode("utf-8")) if raw else str(e)
        raise ApiError(f"HTTP {e.code} calling {url}: {detail}") from e
    except URLError as e:
        raise ApiError(f"Network error: {e}") from e

def create_job(cfg: ApiConfig, payload: Dict[str, Any]) -> Dict[str, Any]:
    return _json_request("POST", f"{cfg.base_url}/v1/jobs", headers={"X-API-Key": cfg.api_key}, body=payload)

def get_job_status(cfg: ApiConfig, job_id: str) -> Dict[str, Any]:
    return _json_request("GET", f"{cfg.base_url}/v1/jobs/{job_id}", headers={"X-API-Key": cfg.api_key})

def list_outputs(cfg: ApiConfig, job_id: str) -> List[Dict[str, Any]]:
    return _json_request("GET", f"{cfg.base_url}/v1/jobs/{job_id}/outputs", headers={"X-API-Key": cfg.api_key})

def wait_for_terminal(cfg: ApiConfig, job_id: str) -> Dict[str, Any]:
    deadline = time.time() + cfg.timeout_seconds
    while time.time() < deadline:
        status = get_job_status(cfg, job_id)
        s = str(status.get("status") or "").lower()
        if s in {"done", "completed", "success"}: return status
        if s in {"failed", "error", "canceled"}: raise ApiError(f"Job failed: {status.get('message')}")
        time.sleep(cfg.poll_seconds)
    raise ApiError("Timed out")

def main():
    cfg = ApiConfig(base_url=BASE_URL, api_key=API_KEY, timeout_seconds=TIMEOUT_SECONDS, poll_seconds=POLL_SECONDS)
    payload = {
        "subject": SUBJECT, "text_center": TEXT_CENTER, "method_no": METHOD_NO, "locale": LOCALE,
        "style": {"custom": bool(CUSTOM_STYLE_TEXT), "style_no": STYLE_NO if not CUSTOM_STYLE_TEXT else None, "custom_text": CUSTOM_STYLE_TEXT}
    }
    job = create_job(cfg, payload)
    job_id = job.get("job_id")
    print(f"Job created: {job_id}")
    wait_for_terminal(cfg, job_id)
    print("Done.")

if __name__ == "__main__":
    main()