1
import requests
2

3
from pixivpy3 import AppPixivAPI
4

5

6

7
_OAUTH_TOKEN_URL = "https://oauth.secure.pixiv.net/auth/token"
8

9
_CLIENT_ID = "MOBrBDS8blbauoSck0ZfDbtuzpyT"
10

11
_CLIENT_SECRET = "lsACyCD94FhDUtGTXi3QzcFE2uU1hqtDaKeqrdwj"
12

13
_UA = "PixivAndroidApp/5.0.234 (Android 11; Pixel 5)"
14

15

16

17
def exchange_by_refresh_token(refresh_token: str):
18

19
    headers = {
20

21
        "User-Agent": _UA,
22

23
        "Referer": "https://app-api.pixiv.net/",
24

25
        "App-OS": "android",
26

27
        "App-OS-Version": "11",
28

29
        "App-Version": "5.0.234",
30

31
    }
32

33
    data = {
34

35
        "client_id": _CLIENT_ID,
36

37
        "client_secret": _CLIENT_SECRET,
38

39
        "grant_type": "refresh_token",
40

41
        "refresh_token": refresh_token,
42

43
        "include_policy": "true",
44

45
    }
46

47
    r = requests.post(_OAUTH_TOKEN_URL, headers=headers, data=data, timeout=30)
48

49
    r.raise_for_status()
50

51
    j = r.json()
52

53
    if "access_token" not in j:
54

55
        raise RuntimeError(f"token 交换失败：{j}")
56

57
    return j["access_token"], j.get("refresh_token", refresh_token)
58

59

60

61
def make_aapi(refresh_token: str) -> AppPixivAPI:
62

63
    at, rt = exchange_by_refresh_token(refresh_token)
64

65
    aapi = AppPixivAPI(timeout=30)
66

67
    aapi.set_auth(at, rt)
68

69
    try: aapi.set_accept_language("zh-CN")
70

71
    except: pass
72

73
    return aapi

1
import random
2

3

4

5
def extract_pages(ill: dict) -> list[str]:
6

7
    out = []
8

9
    if ill.get("meta_pages"):
10

11
        for p in ill["meta_pages"]:
12

13
            img = p.get("image_urls") or {}
14

15
            u = img.get("original") or img.get("large") or img.get("medium")
16

17
            if u: out.append(u)
18

19
    if not out:
20

21
        if ill.get("meta_single_page", {}).get("original_image_url"):
22

23
            out.append(ill["meta_single_page"]["original_image_url"])
24

25
        else:
26

27
            img = ill.get("image_urls") or {}
28

29
            u = img.get("original") or img.get("large") or img.get("medium")
30

31
            if u: out.append(u)
32

33
    return out
34

35

36

37
def random_one_from_uid(aapi, uid: int, max_jumps: int = 30) -> str | None:
38

39
    res = aapi.user_illusts(uid)
40

41
    if not res or not getattr(res, "illusts", None): return None
42

43
    jumps = random.randint(0, max_jumps)
44

45
    cur = res
46

47
    for _ in range(jumps):
48

49
        if not getattr(cur, "next_url", None): break
50

51
        qs = aapi.parse_qs(cur.next_url)
52

53
        cur = aapi.user_illusts(**qs)
54

55
        if not cur or not getattr(cur, "illusts", None): break
56

57
    cand = getattr(cur, "illusts", []) or []
58

59
    if not cand: return None
60

61
    pages = extract_pages(random.choice(cand))
62

63
    return random.choice(pages) if pages else None
64

65

66

67
def random_one_from_tag(aapi, tag: str, max_jumps: int = 30) -> str | None:
68

69
    res = aapi.search_illust(tag, search_target="partial_match_for_tags")
70

71
    if not res or not getattr(res, "illusts", None): return None
72

73
    jumps = random.randint(0, max_jumps)
74

75
    cur = res
76

77
    for _ in range(jumps):
78

79
        if not getattr(cur, "next_url", None): break
80

81
        qs = aapi.parse_qs(cur.next_url)
82

83
        cur = aapi.search_illust(**qs)
84

85
        if not cur or not getattr(cur, "illusts", None): break
86

87
    cand = getattr(cur, "illusts", []) or []
88

89
    if not cand: return None
90

91
    pages = extract_pages(random.choice(cand))
92

93
    return random.choice(pages) if pages else None

1
async def notify_owner(bot, owner_qq: int, text: str, event=None):
2

3
    try:
4

5
        if hasattr(bot, "send_private"):
6

7
            await bot.send_private(owner_qq, f"[setu告警] {text}"); return
8

9
    except Exception:
10

11
        pass
12

13
    if event is not None:
14

15
        await bot.send(event, f"[setu告警] @{owner_qq} {text}")
16

17

18

19
async def get_by_pid_uid_tag_with_fallback(session, bot, event, owner_qq, refresh_token,
20

21
                                           pid: int|None, uid: int|None, tag: str|None,
22

23
                                           fetch_from_api, fetch_from_db):
24

25
    try:
26

27
        aapi = await asyncio.to_thread(make_aapi, refresh_token)
28

29
        if pid is not None:
30

31
            det = await asyncio.to_thread(aapi.illust_detail, pid)
32

33
            ill = det.illust
34

35
            urls = extract_pages(ill) if ill else []
36

37
        elif uid is not None:
38

39
            u = await asyncio.to_thread(random_one_from_uid, aapi, uid)
40

41
            urls = [u] if u else []
42

43
        elif tag:
44

45
            u = await asyncio.to_thread(random_one_from_tag, aapi, tag)
46

47
            urls = [u] if u else []
48

49
        else:
50

51
            urls = []
52

53
        if urls: return urls
54

55
        await notify_owner(bot, owner_qq, "官方为空，回退 API", event)
56

57
    except Exception as e:
58

59
        await notify_owner(bot, owner_qq, f"官方异常：{e}，回退 API", event)
60

61

62

63
    # API
64

65
    try:
66

67
        api_urls = await fetch_from_api(session, [] if pid or uid else ([f"tag={tag}"] if tag else []))
68

69
        if api_urls: return api_urls[:1]
70

71
        await notify_owner(bot, owner_qq, "API 为空，回退 DB", event)
72

73
    except Exception as e:
74

75
        await notify_owner(bot, owner_qq, f"API 异常：{e}，回退 DB", event)
76

77

78

79
    # DB
80

81
    try:
82

83
        db_urls = await fetch_from_db(1, None, tag, uid)
84

85
        return db_urls[:1] if db_urls else []
86

87
    except Exception as e:
88

89
        await notify_owner(bot, owner_qq, f"DB 异常：{e}，最终失败", event)
90

91
        return []