67 lines
2.5 KiB
Python
67 lines
2.5 KiB
Python
import os, requests
|
|
from typing import Any, Dict, List, Tuple, Optional
|
|
from utils.boxapi_log import log_boxapi_response
|
|
|
|
BOXAPI_BASE = os.environ.get("BOXAPI_BASE", "https://boxapi.ir")
|
|
BOXAPI_KEY = os.environ.get("BOXAPI_KEY", "") # put your key in .env
|
|
|
|
class BoxAPIError(Exception):
|
|
pass
|
|
|
|
def get_media(username: str, count: int = 5) -> Tuple[List[Dict[str, Any]], str]:
|
|
"""
|
|
Returns (items, log_path)
|
|
items is the array under response.body.items
|
|
"""
|
|
url = f"{BOXAPI_BASE}/api/instagram/user/get_media"
|
|
payload = {"username": username, "count": count}
|
|
headers = {"Authorization": f"Bearer {BOXAPI_KEY}"} if BOXAPI_KEY else {}
|
|
|
|
resp = None
|
|
resp_json: Optional[Dict[str, Any]] = None
|
|
status_code: Optional[int] = None
|
|
|
|
try:
|
|
resp = requests.post(url, json=payload, headers=headers, timeout=60)
|
|
status_code = resp.status_code
|
|
resp.raise_for_status()
|
|
resp_json = resp.json()
|
|
except Exception as e:
|
|
log_path = log_boxapi_response(username, payload, resp_json, status_code, note=f"HTTP/parse error: {e}")
|
|
raise BoxAPIError(f"BoxAPI request failed ({e}). See log: {log_path}")
|
|
|
|
# ✅ Real structure according to your sample
|
|
# {
|
|
# "status": "done",
|
|
# "response": {
|
|
# "status_code": 200,
|
|
# "content_type": "application/json",
|
|
# "body": {
|
|
# "items": [ ... ]
|
|
# }
|
|
# }
|
|
# }
|
|
try:
|
|
top_status = resp_json.get("status")
|
|
inner = resp_json.get("response") or {}
|
|
inner_code = inner.get("status_code")
|
|
body = inner.get("body") or {}
|
|
items = body.get("items") or []
|
|
|
|
log_path = log_boxapi_response(username, payload, resp_json, inner_code, note="OK")
|
|
|
|
if top_status != "done" or inner_code != 200 or not isinstance(items, list):
|
|
raise ValueError(f"Unexpected shape: status={top_status}, inner_code={inner_code}, items_type={type(items)}")
|
|
return items, log_path
|
|
except Exception as e:
|
|
log_path = log_boxapi_response(username, payload, resp_json, status_code, note=f"schema error: {e}")
|
|
raise BoxAPIError(f"Unexpected response format from BoxAPI. See log: {log_path}")
|
|
|
|
# Legacy function for backward compatibility
|
|
def fetch_boxapi_posts(username: str, count: int = 5) -> List[Dict]:
|
|
try:
|
|
items, _ = get_media(username, count)
|
|
return items
|
|
except BoxAPIError as e:
|
|
print(f"❌ Failed to fetch posts from BoxAPI: {e}")
|
|
return [] |