main
1#!/usr/bin/env python
2# -*- coding: utf-8 -*-
3from typing import Literal
4
5from glom import glom
6from pyrogram.client import Client
7from pyrogram.types import Message
8
9from bridge.social import send_to_social_media_bridge
10from config import PREFIX, PROXY, TOKEN, cache
11from messages.progress import modify_progress
12from messages.sender import send2tg
13from networking import download_file, hx_req
14from utils import seconds_to_time, zhcn
15
16
17async def preview_spotify(
18 client: Client,
19 message: Message,
20 resource: Literal["track", "album", "artist", "playlist"],
21 spotify_id: str,
22 url: str,
23 **kwargs,
24):
25 """Preview spotify info in the message."""
26 if kwargs.get("show_progress") and "progress" not in kwargs:
27 res = await send2tg(client, message, texts=f"🔗正在解析Spotify链接\n{url}", **kwargs)
28 kwargs["progress"] = res[0]
29 if resource == "track":
30 await preview_track(client, message, url, spotify_id, **kwargs)
31 elif resource == "album":
32 await preview_album(client, message, url, spotify_id, **kwargs)
33
34
35async def preview_track(client: Client, message: Message, url: str, spotify_id: str, **kwargs):
36 api = f"https://api.spotify.com/v1/tracks/{spotify_id}"
37 access_token = await get_access_token()
38 headers = {"Authorization": f"Bearer {access_token}"}
39 resp = await hx_req(api, headers=headers, proxy=PROXY.SPOTIFY, check_kv={"id": spotify_id})
40
41 status = f"🎧歌曲: [{resp['name']}]({url})\n"
42 if artists := resp.get("artists", []):
43 for artist in artists:
44 artist_url = glom(artist, "external_urls.spotify", default=url)
45 status += f"🎙歌手: [{artist['name']}]({artist_url})\n"
46 if duration_ms := resp.get("duration_ms"):
47 status += f"⏱时长: {seconds_to_time(duration_ms / 1000)}\n"
48 if album_name := glom(resp, "album.name", default=None):
49 album_url = glom(resp, "album.external_urls.spotify", default=url)
50 status += f"💿专辑: [{album_name}]({album_url})\n"
51 if release_date := glom(resp, "album.release_date", default=None):
52 status += f"🗓日期: {release_date}\n"
53
54 texts_to_other_bot = resp["name"]
55 if artist := glom(resp, "artists.0.name", default=None):
56 texts_to_other_bot = f"{artist} - {resp['name']}"
57 texts_to_other_bot = zhcn(texts_to_other_bot)
58 # add warning
59 status_warning = "\n⚠️我无法直接下载歌曲\n正在发送歌曲信息给第三方Bot:\n@Music163bot"
60 caption_warning = f"\n⚠️本歌曲由 @Music163bot 通过以下关键词搜索获得\n`{texts_to_other_bot}`\n"
61 caption_warning += "如不准确可私聊以下Bot自行搜索:\n @Music163bot @VmomoVBot\n"
62 caption_warning += f"您也发送以下命令在YouTube上查找:\n`{PREFIX.SEARCH_YOUTUBE} {texts_to_other_bot}`"
63 await modify_progress(text=status + status_warning, **kwargs)
64
65 kwargs |= {
66 "send_from_user": "", # disable @send_user
67 "caption": status + caption_warning,
68 "prefix-Music163bot": "/music ",
69 "target_mid": message.id, # record the trigger msg_id as target_mid in kwargs
70 }
71 await send_to_social_media_bridge(client, message, texts_to_other_bot, **kwargs)
72
73
74async def preview_album(client: Client, message: Message, url: str, spotify_id: str, **kwargs):
75 api = f"https://api.spotify.com/v1/albums/{spotify_id}"
76 access_token = await get_access_token()
77 headers = {"Authorization": f"Bearer {access_token}"}
78 resp = await hx_req(api, headers=headers, proxy=PROXY.SPOTIFY, check_kv={"id": spotify_id})
79 status = f"💿专辑: [{resp['name']}]({url})\n"
80 if artists := resp.get("artists", []):
81 for artist in artists:
82 artist_url = glom(artist, "external_urls.spotify", default=url)
83 status += f"🎙歌手: [{artist['name']}]({artist_url})\n"
84 if release_date := resp.get("release_date"):
85 status += f"🗓日期: {release_date}\n"
86 if tracks := glom(resp, "tracks.items", default=[]):
87 status += "🎧歌曲:\n"
88 for idx, track in enumerate(tracks):
89 track_url = glom(track, "external_urls.spotify", default=url)
90 status += f"{idx + 1}. [{track['name']}]({track_url}) ({seconds_to_time(track['duration_ms'] / 1000)})\n"
91 cover_url = glom(resp, "images.0.url", default="")
92 media = [{"photo": await download_file(cover_url, suffix=".jpg", proxy=PROXY.SPOTIFY)}] if cover_url else []
93 kwargs["send_from_user"] = "" # disable @send_user
94 await send2tg(client, message, texts=status, media=media, **kwargs)
95 await modify_progress(del_status=True, **kwargs)
96
97
98@cache.memoize(ttl=3600)
99async def get_access_token():
100 api = "https://accounts.spotify.com/api/token"
101 headers = {"Content-Type": "application/x-www-form-urlencoded"}
102 data = f"grant_type=client_credentials&client_id={TOKEN.SPOTIFY_CLIENT_ID}&client_secret={TOKEN.SPOTIFY_CLIENT_SECRET}"
103 resp = await hx_req(api, method="POST", proxy=PROXY.SPOTIFY, content_data=data, headers=headers, check_keys=["access_token"])
104 return resp["access_token"]