137 lines
4.5 KiB
Python
137 lines
4.5 KiB
Python
# utils/boxapi_client.py
|
||
import os
|
||
import requests
|
||
from typing import Any, Dict, List, Tuple, Optional
|
||
from utils.boxapi_log import log_boxapi_response # keep your existing logger
|
||
|
||
BOXAPI_USERNAME = os.environ.get("BOXAPI_USERNAME", "")
|
||
BOXAPI_PASSWORD = os.environ.get("BOXAPI_PASSWORD", "")
|
||
BOXAPI_ENDPOINT = os.environ.get(
|
||
"BOXAPI_ENDPOINT",
|
||
"https://boxapi.ir/api/instagram/user/get_media_by_username",
|
||
)
|
||
BOXAPI_DEBUG = os.getenv("BOXAPI_DEBUG", "0") == "1"
|
||
|
||
|
||
class BoxAPIError(Exception):
|
||
pass
|
||
|
||
|
||
def _dprint(msg: str) -> None:
|
||
if BOXAPI_DEBUG:
|
||
print(f"[BOXAPI] {msg}")
|
||
|
||
|
||
def get_media_page(
|
||
username: str,
|
||
count: int = 12,
|
||
max_id: Optional[str] = None,
|
||
) -> Tuple[List[Dict[str, Any]], str, Optional[str]]:
|
||
"""
|
||
Fetch ONE page from BoxAPI.
|
||
Returns: (items, log_path, next_max_id)
|
||
"""
|
||
if not BOXAPI_USERNAME or not BOXAPI_PASSWORD:
|
||
raise BoxAPIError("Missing BOXAPI_USERNAME/BOXAPI_PASSWORD in environment.")
|
||
|
||
per_req = max(1, min(int(count or 1), 12)) # BoxAPI cap per request
|
||
payload: Dict[str, Any] = {"username": username, "count": per_req}
|
||
if max_id:
|
||
payload["max_id"] = max_id
|
||
|
||
_dprint(f"→ POST {BOXAPI_ENDPOINT} payload={payload}")
|
||
|
||
resp_json: Optional[Dict[str, Any]] = None
|
||
resp_text: Optional[str] = None
|
||
status_code: Optional[int] = None
|
||
|
||
try:
|
||
resp = requests.post(
|
||
BOXAPI_ENDPOINT,
|
||
json=payload,
|
||
auth=(BOXAPI_USERNAME, BOXAPI_PASSWORD),
|
||
timeout=60,
|
||
)
|
||
status_code = resp.status_code
|
||
resp_text = resp.text
|
||
|
||
try:
|
||
resp_json = resp.json()
|
||
except Exception:
|
||
# Non-JSON – log and raise
|
||
log_path = log_boxapi_response(
|
||
username, payload, status_code, response_text=resp_text,
|
||
note="Non-JSON response (error)"
|
||
)
|
||
raise BoxAPIError(f"BoxAPI non-JSON response. See log: {log_path}")
|
||
|
||
# Expected shape:
|
||
# {"status":"done", "response":{"status_code":200, "body":{"items":[...], "next_max_id":"..."}}}
|
||
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 []
|
||
next_max_id = body.get("next_max_id")
|
||
|
||
log_path = log_boxapi_response(
|
||
username, payload, inner_code, response_json=resp_json, note="OK"
|
||
)
|
||
|
||
if top_status != "done" or inner_code != 200 or not isinstance(items, list):
|
||
snap = {
|
||
"top_status": top_status,
|
||
"inner_code": inner_code,
|
||
"has_body": isinstance(body, dict),
|
||
"items_type": str(type(items)),
|
||
"items_len": len(items) if isinstance(items, list) else None,
|
||
}
|
||
raise BoxAPIError(f"Unexpected response format from BoxAPI ({snap}). See log: {log_path}")
|
||
|
||
_dprint(f"← items={len(items)} next_max_id={next_max_id!r}")
|
||
return items, log_path, next_max_id
|
||
|
||
except BoxAPIError:
|
||
raise
|
||
except Exception as e:
|
||
log_path = log_boxapi_response(
|
||
username, payload, status_code,
|
||
response_json=resp_json, response_text=resp_text,
|
||
note=f"HTTP/parse error (error): {e}"
|
||
)
|
||
raise BoxAPIError(f"BoxAPI request failed ({e}). See log: {log_path}")
|
||
|
||
|
||
def get_media(username: str, count: int = 12) -> Tuple[List[Dict[str, Any]], str]:
|
||
"""
|
||
Backward-compatible helper: fetch ONE latest page without pagination.
|
||
Returns: (items, log_path)
|
||
"""
|
||
items, log_path, _next = get_media_page(username=username, count=count, max_id=None)
|
||
return items, log_path
|
||
|
||
|
||
def fetch_boxapi_posts(username: str, count: int = 30) -> List[Dict[str, Any]]:
|
||
"""
|
||
Backward-compatible helper that PAGES until it collects `count` items (or exhausts).
|
||
"""
|
||
try:
|
||
target = max(1, int(count))
|
||
collected: List[Dict[str, Any]] = []
|
||
next_id: Optional[str] = None
|
||
|
||
while len(collected) < target:
|
||
need = min(12, target - len(collected))
|
||
items, _log, next_id = get_media_page(username=username, count=need, max_id=next_id)
|
||
if not items:
|
||
break
|
||
collected.extend(items)
|
||
if not next_id:
|
||
break
|
||
|
||
return collected[:target]
|
||
|
||
except BoxAPIError as e:
|
||
print(f"❌ Failed to fetch posts from BoxAPI: {e}")
|
||
return []
|