mirror of
https://github.com/Lost-MSth/Arcaea-server.git
synced 2026-02-10 01:37:27 +08:00
[Enhance] Restrict download by user's purchase
- An option `DOWNLOAD_FORBID_WHEN_NO_ITEM` has been added to the config file to make that users cannot download the songs' files if they has not bought them when the `songlist` file exists. (Experimental) #128
This commit is contained in:
@@ -48,6 +48,8 @@ class Config:
|
||||
DOWNLOAD_TIMES_LIMIT = 3000
|
||||
DOWNLOAD_TIME_GAP_LIMIT = 1000
|
||||
|
||||
DOWNLOAD_FORBID_WHEN_NO_ITEM = False
|
||||
|
||||
LOGIN_DEVICE_NUMBER_LIMIT = 1
|
||||
ALLOW_LOGIN_SAME_DEVICE = False
|
||||
ALLOW_BAN_MULTIDEVICE_USER_AUTO = True
|
||||
|
||||
@@ -21,6 +21,9 @@ class Constant:
|
||||
|
||||
WORLD_VALUE_NAME_ENUM = ['frag', 'prog', 'over']
|
||||
|
||||
FREE_PACK_NAME = 'base'
|
||||
SINGLE_PACK_NAME = 'single'
|
||||
|
||||
ETO_UNCAP_BONUS_PROGRESS = 7
|
||||
LUNA_UNCAP_BONUS_PROGRESS = 7
|
||||
AYU_UNCAP_BONUS_PROGRESS = 5
|
||||
|
||||
@@ -26,6 +26,8 @@ class SonglistParser:
|
||||
|
||||
FILE_NAMES = ['0.aff', '1.aff', '2.aff', '3.aff',
|
||||
'base.ogg', '3.ogg', 'video.mp4', 'video_audio.ogg']
|
||||
|
||||
has_songlist = False
|
||||
songs: dict = {} # {song_id: value, ...}
|
||||
# value: bit 76543210
|
||||
# 7: video_audio.ogg
|
||||
@@ -37,6 +39,10 @@ class SonglistParser:
|
||||
# 1: 1.aff
|
||||
# 0: 0.aff
|
||||
|
||||
pack_info: 'dict[str, set]' = {} # {pack_id: {song_id, ...}, ...}
|
||||
free_songs: set = set() # {song_id, ...}
|
||||
world_songs: set = set() # {world_song_id, ...}
|
||||
|
||||
def __init__(self, path=Constant.SONGLIST_FILE_PATH) -> None:
|
||||
self.path = path
|
||||
self.data: list = []
|
||||
@@ -54,10 +60,31 @@ class SonglistParser:
|
||||
return True
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def get_user_unlocks(user) -> set:
|
||||
'''user: UserInfo类或子类的实例'''
|
||||
x = SonglistParser
|
||||
if user is None:
|
||||
return set()
|
||||
|
||||
r = set()
|
||||
for i in user.packs:
|
||||
if i in x.pack_info:
|
||||
r.update(x.pack_info[i])
|
||||
|
||||
if Constant.SINGLE_PACK_NAME in x.pack_info:
|
||||
r.update(x.pack_info[Constant.SINGLE_PACK_NAME]
|
||||
& set(user.singles))
|
||||
r.update(set(i if i[-1] != '3' else i[:-1]
|
||||
for i in (x.world_songs & set(user.world_songs))))
|
||||
r.update(x.free_songs)
|
||||
|
||||
return r
|
||||
|
||||
def parse_one(self, song: dict) -> dict:
|
||||
'''解析单个歌曲'''
|
||||
if not 'id' in song:
|
||||
return None
|
||||
return {}
|
||||
r = 0
|
||||
if 'remote_dl' in song and song['remote_dl']:
|
||||
r |= 16
|
||||
@@ -76,21 +103,42 @@ class SonglistParser:
|
||||
r |= 128
|
||||
return {song['id']: r}
|
||||
|
||||
def parse_one_unlock(self, song: dict) -> None:
|
||||
'''解析单个歌曲解锁方式'''
|
||||
if not 'id' in song or not 'set' in song or not 'purchase' in song:
|
||||
return {}
|
||||
x = SonglistParser
|
||||
if Constant.FREE_PACK_NAME == song['set']:
|
||||
if any(i['ratingClass'] == 3 for i in song.get('difficulties', [])):
|
||||
x.world_songs.add(song['id'] + '3')
|
||||
x.free_songs.add(song['id'])
|
||||
return None
|
||||
|
||||
if song.get('world_unlock', False):
|
||||
x.world_songs.add(song['id'])
|
||||
|
||||
if song['purchase'] == '':
|
||||
return None
|
||||
|
||||
x.pack_info.setdefault(song['set'], set()).add(song['id'])
|
||||
|
||||
def parse(self) -> None:
|
||||
'''解析songlist文件'''
|
||||
if not os.path.isfile(self.path):
|
||||
return
|
||||
with open(self.path, 'r', encoding='utf-8') as f:
|
||||
self.data = loads(f.read()).get('songs', [])
|
||||
self.has_songlist = True
|
||||
for x in self.data:
|
||||
self.songs.update(self.parse_one(x))
|
||||
self.parse_one_unlock(x)
|
||||
|
||||
|
||||
class UserDownload:
|
||||
'''
|
||||
用户下载类
|
||||
|
||||
properties: `user` - `User`类或子类的实例
|
||||
properties: `user` - `UserInfo`类或子类的实例
|
||||
'''
|
||||
|
||||
limiter = ArcLimiter(
|
||||
@@ -198,6 +246,10 @@ class DownloadList(UserDownload):
|
||||
DownloadList.get_one_song_file_names.cache_clear()
|
||||
DownloadList.get_all_song_ids.cache_clear()
|
||||
SonglistParser.songs = {}
|
||||
SonglistParser.pack_info = {}
|
||||
SonglistParser.free_songs = set()
|
||||
SonglistParser.world_songs = set()
|
||||
SonglistParser.has_songlist = False
|
||||
|
||||
def clear_download_token(self) -> None:
|
||||
'''清除过期下载链接'''
|
||||
@@ -277,9 +329,19 @@ class DownloadList(UserDownload):
|
||||
|
||||
if not self.song_ids:
|
||||
self.song_ids = self.get_all_song_ids()
|
||||
if Config.DOWNLOAD_FORBID_WHEN_NO_ITEM and SonglistParser.has_songlist:
|
||||
# 没有歌曲时不允许下载
|
||||
self.song_ids = list(SonglistParser.get_user_unlocks(
|
||||
self.user) & set(self.song_ids))
|
||||
|
||||
for i in self.song_ids:
|
||||
self.add_one_song(i)
|
||||
else:
|
||||
if Config.DOWNLOAD_FORBID_WHEN_NO_ITEM and SonglistParser.has_songlist:
|
||||
# 没有歌曲时不允许下载
|
||||
self.song_ids = list(SonglistParser.get_user_unlocks(
|
||||
self.user) & set(self.song_ids))
|
||||
|
||||
for i in self.song_ids:
|
||||
if os.path.isdir(os.path.join(Constant.SONG_FILE_FOLDER_PATH, i)):
|
||||
self.add_one_song(i)
|
||||
|
||||
Reference in New Issue
Block a user