File size: 2,496 Bytes
607c0f5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 | import os
from typing import Any, Dict, List, Optional
import requests
ROOT_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
DEFAULT_ENDPOINT = "https://q6-p.hf.space"
REQUEST_TIMEOUT = 300
def read_dotenv_value(path: str, key: str) -> Optional[str]:
try:
with open(path, "r") as env_file:
for line in env_file:
line = line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
k, v = line.split("=", 1)
if k == key:
return v
except FileNotFoundError:
return None
return None
def get_phpsessid() -> str:
phpsessid = os.getenv("PHPSESSID")
if phpsessid:
return phpsessid
env_path = os.path.join(ROOT_DIR, ".env")
phpsessid = read_dotenv_value(env_path, "PHPSESSID")
if phpsessid:
return phpsessid
raise RuntimeError("PHPSESSID is not set in the environment or .env")
def get_endpoint() -> str:
endpoint = os.getenv("PIXIF_ENDPOINT")
if endpoint:
return endpoint.rstrip("/")
env_path = os.path.join(ROOT_DIR, ".env")
endpoint = read_dotenv_value(env_path, "PIXIF_ENDPOINT")
if endpoint:
return endpoint.rstrip("/")
return DEFAULT_ENDPOINT
def fetch_search(
url: str,
pages: int,
ai_only: bool,
real_only: bool,
phpsessid: str,
endpoint: str,
) -> Dict[str, Any]:
payload = {
"url": url,
"pages": pages,
"ai_only": ai_only,
"real_only": real_only,
"phpsessid": phpsessid,
}
response = requests.post(f"{endpoint}/search", json=payload, timeout=REQUEST_TIMEOUT)
response.raise_for_status()
data = response.json()
if isinstance(data, dict) and data.get("error"):
raise RuntimeError(str(data.get("error")))
return data
def fetch_users(user_ids: List[int], phpsessid: str, endpoint: str) -> List[Dict[str, Any]]:
payload = {"user_ids": user_ids, "phpsessid": phpsessid}
response = requests.post(f"{endpoint}/users", json=payload, timeout=REQUEST_TIMEOUT)
response.raise_for_status()
data = response.json()
if isinstance(data, dict) and data.get("error"):
raise RuntimeError(str(data.get("error")))
if isinstance(data, dict) and "users" in data:
data = data["users"]
if not isinstance(data, list):
raise RuntimeError("Unexpected response from users endpoint.")
return data
|