- Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathnico.py
More file actions
Latest commit
432 lines (387 loc) · 19.7 KB
/
Copy pathnico.py
File metadata and controls
432 lines (387 loc) · 19.7 KB
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
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
importconcurrent.futures
importjson
importre
fromdatetimeimportdatetime
fromsubprocessimportrun
fromurllib.parseimporturljoin, urlparse
fromurllib.requestimportgetproxies
frompathlibimportPath
importwebsocket
fromrich.consoleimportConsole
fromrich.tableimportTable
fromutilimportdownload, dump_json, get, MyTime, requests_retry_session, safeify, to_jp_time, load_cookie
fromproto.dwango.nicolive.chat.service.edgeimportpayload_pb2aschat
importgoogle.protobuf.json_format
console=Console()
print=console.print
# based on https://github.com/rinsuki-lab/ndgr-reader/blob/main/src/protobuf-stream-reader.ts
defread_protobuf_message(data):
offset=0
result=0
i=0
whileTrue:
ifoffset>=len(data):
returnNone
current=data[offset]
result|= (current&0x7F) <<i
offset+=1
i+=7
ifnot (current&0x80):
break
ifoffset+result>len(data):
returnNone
returndata[offset:offset+result]
classNicoDownloader():
HEADERS= {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/112.0.0.0 Safari/537.36'}
def__init__(self, cookies, proxy=None, save_dir=None):
defvalidate_cookie(cookies):
forcookieincookies:
ifcookie.name=='user_session'andcookie.value:
print('Find user_session in cookie:', cookie.value)
break
else:
print(f'WARN: Cannot find user_session in cookie. You\'re probably not logged in.')
# exit(1) # make it non-fatal
ifisinstance(cookies, str) andcookies.lower() in ['chrome', 'firefox', 'edge']:
cookies=load_cookie(cookies+'/nicovideo.jp')
validate_cookie(cookies)
elifisinstance(cookies, str) andcookies.startswith('user_session_'):
cookies= {'user_session': cookies}
elifPath(cookies).exists():
print(f'Loading cookies from file {cookies}...')
cookies=load_cookie(cookies)
validate_cookie(cookies)
else:
print('ERROR: Invalid cookie source. Please provide a browser name, a cookie value, or a Netscape-style cookie file.')
exit(1)
self.session=requests_retry_session()
self.session.cookies.update(cookies)
self.session.headers.update(self.HEADERS)
# proxy settings
ifnotproxyorproxy.lower() =='none':
# it's recommended to set proxy to '' instead of None, otherwise
# some redirected requests may not actually use the proxy settings.
# see https://github.com/psf/requests/issues/6153
proxy=''
elifproxy=='auto':
sys_proxies=getproxies()
ifproxy:=sys_proxies.get('http', ''): # if failed, proxy would be ''
print(f'INFO: Automatically use system proxy {proxy}')
# I don't think system proxy would be missing scheme ever, but just in case
ifproxyand'://'notinproxy:
print('WARN: Proxy is missing scheme. Assuming http://')
proxy=f'http://{proxy}'
# setting self.session.proxies alone isn't enough; because requests will
# prioritize system proxy over session.proxies.
# you have to set proxy at request level to override system proxy.
# see https://github.com/psf/requests/issues/2018
# so, we have to disable system proxy by setting trust_env to False first.
self.session.trust_env=False
self.session.proxies= {'http': proxy, 'https': proxy}
# this is for websocket connection/minyami download
self.proxy=proxy
self.save_dir=Path(save_dir) ifsave_direlsePath.cwd()
def_parse_url_or_video_id(self, url_or_video_id):
ifm:=re.search(r'/watch/([^?&]+)', url_or_video_id):
video_id=m[1]
else:
video_id=url_or_video_id
ifvideo_id.startswith('lv'):
url=f'https://live.nicovideo.jp/watch/{video_id}'
video_type='live'
else:
url=f'https://www.nicovideo.jp/watch/{video_id}'
video_type='video'
returnvideo_id, url, video_type
defcreate_ws(self, url):
host, port, type_=None, None, None
ifself.proxy:
parsed=urlparse(self.proxy)
host, port, type_=parsed.hostname, parsed.port, parsed.scheme
returnwebsocket.create_connection(url, header=self.HEADERS, http_proxy_host=host, http_proxy_port=port, proxy_type=type_)
deffetch_page(self, url):
soup=get(url, session=self.session)
live_data=json.loads(soup.select_one('#embedded-data')['data-props'])
returnlive_data
defdownload_comments_native(self, message_server_info, output):
view_uri=message_server_info['data']['viewUri']
#"vposBaseTime": "2024-09-25T21:50:00+09:00",
vpos_base_time_dt=datetime.strptime(message_server_info['data']['vposBaseTime'], '%Y-%m-%dT%H:%M:%S%z')
vpos_base_time_epoch=int(vpos_base_time_dt.timestamp())
print(f'vpos Base time: {vpos_base_time_dt} ({vpos_base_time_epoch})')
at='now'
backward_api_uri=None
whileTrue:
url=f'{view_uri}?&at={at}'
print(f'Fetch {url}')
r=self.session.get(url, timeout=30)
message=read_protobuf_message(r.content)
chunked_entry=chat.ChunkedEntry()
chunked_entry.ParseFromString(message)
ifchunked_entry.HasField('next'):
at=chunked_entry.next.at
elifchunked_entry.HasField('backward'):
backward_api_uri=chunked_entry.backward.segment.uri
break
messages= []
whileTrue:
print(f'Fetch {backward_api_uri}')
r2=self.session.get(backward_api_uri, timeout=30)
packed_segment=chat.PackedSegment()
packed_segment.ParseFromString(r2.content)
# prepend to messages
messages= [messageformessageinpacked_segment.messages] +messages
ifpacked_segment.HasField('next'):
backward_api_uri=packed_segment.next.uri
else:
break
print(f'Find {len(messages)} messages.')
dump_json([google.protobuf.json_format.MessageToDict(message) formessageinmessages], output)
# TODO: convert the json to a format that is compatible with nicoxml2ass
defdownload_timeshift(self, url_or_video_id, info_only=False, comments='no', verbose=False, dump=False, auto_reserve=False, simulate=False):
video_id, url, video_type=self._parse_url_or_video_id(url_or_video_id)
return_value= {
'id': video_id,
'url': url,
'type': video_type,
}
# download video type is not implemented yet
ifvideo_type!='live':
print('ERROR: Download video type is not implemented yet.')
returnreturn_value
live_data=self.fetch_page(url)
title=live_data['program']['title']
begin_time_epoch=live_data["program"]["beginTime"]
end_time_epoch=live_data["program"]["endTime"]
begin_time_dt=to_jp_time(datetime.fromtimestamp(begin_time_epoch))
end_time_dt=to_jp_time(datetime.fromtimestamp(end_time_epoch))
date=begin_time_dt.strftime('%y%m%d')
max_quality=live_data['program']['stream']['maxQuality']
filename=safeify(f"{date}{title}_{video_id}")
t=Table(show_header=False, show_lines=True)
t.add_column('Desc.', style='bold green')
t.add_column('Value')
t.add_row('Video ID', video_id)
t.add_row('Title', title)
t.add_row('Start time', MyTime(begin_time_dt).jst("pretty") +" (JST)")
t.add_row('Max quality', max_quality)
t.add_row('Filename', filename)
print(t)
return_value.update({
'title': title,
'begin_time': begin_time_dt,
'end_time': end_time_dt,
'short_date': date,
'max_quality': max_quality,
'filename': filename,
'info': live_data
})
ifdump:
dump_json(live_data, self.save_dir/f'{filename}.info.json')
ifinfo_only:
returnreturn_value
# check video availability
# use while, this way when we reserve/activate timeshift ticket, we can refetch live_data and recheck
# to see if there is any other errors
whilenotlive_data['site']['relive'].get('webSocketUrl', None):
assertlive_data['userProgramWatch']['canWatch'] ==False
# return if isCountryRestrictionTarget is true
iflive_data['userProgramWatch'].get('isCountryRestrictionTarget', False):
print('ERROR: This video is not available in your country.')
returnreturn_value
print(f'WARN: You don\'t have or have not activated the timeshift ticket. Reason:\n{live_data["userProgramWatch"]}')
ifauto_reserveorinput('Do you want to reserve/activate it now? Y/[N] ').lower() =='y':
print('Reserving...')
# POST = reserve, PATCH = activate/use
reservation_url=f'https://live2.nicovideo.jp/api/v2/programs/{video_id}/timeshift/reservation'
r=self.session.post(reservation_url)
print('Tried POST, response:', r.status_code)
r=self.session.patch(reservation_url)
print('Tried PATCH, response:', r.status_code)
ifr.status_code!=200:
print('Reserving or activating failed. Please try reserving it manually on the webpage.')
returnreturn_value
# refetch live_data
old_user_program_watch=live_data['userProgramWatch']
live_data=self.fetch_page(url)
iflive_data['userProgramWatch'] ==old_user_program_watch:
print('WARN: live_data did not change after reserving/activating. Something must be wrong.')
print('Aborted.')
returnreturn_value
# back to the beginning of the loop
else:
print("Aborted.")
returnreturn_value
# Add warning if it's trial only
iflive_data['programWatch']['condition'].get('payment') =='Ticket'andnotlive_data['userProgramWatch']['payment']['hasTicket']:
# you can always download full comments, so no need to check if comments == 'only'
ifcomments=='only':
pass
ifinput('WARN: This timeshift requires a ticket but you don\'t have one. '
'The video will only have the trial part, and be black afterwards. '
'Do you want to continue? Y/[N] ').lower() !='y':
returnreturn_value
ws_url=live_data['site']['relive']['webSocketUrl']
audience_token=re.search(r'audience_token=(.+)', ws_url)[1]
print(f'WS url is {ws_url}')
print('Creating websocket connection...')
# websocket.enableTrace(True)
ws=self.create_ws(ws_url)
verboseandprint("Sent startWatching")
start_watching_payload= {
"type": "startWatching",
"data": {
"stream": {
"quality": max_quality,
"protocol": "hls",
"latency": "low",
"chasePlay": False,
'accessRightMethod': 'single_cookie'
},
"room": {
"protocol": "webSocket",
"commentable": True
},
"reconnect": False
}
}
verboseandprint('Payload:', start_watching_payload)
ws.send(json.dumps(start_watching_payload))
stream_info=None
message_server_info=None
whileTrue:
verboseandprint("Receiving...")
result=ws.recv()
verboseandprint("Received '%s'"%result)
data=json.loads(result)
ifdata['type'] =='stream':
stream_info=data
elifdata['type'] =='messageServer':
message_server_info=data
ifstream_infoandmessage_server_info:
print('Got all the info we needed. Close WS connection.')
break
ws.close()
ifdump:
dump_json(stream_info, self.save_dir/f'{filename}.streaminfo.json')
dump_json(message_server_info, self.save_dir/f'{filename}.msgserverinfo.json')
return_value.update({
'stream_info': stream_info,
'message_server_info': message_server_info
})
ex=concurrent.futures.ThreadPoolExecutor(max_workers=1)
ifnotsimulateandcommentsin ['yes', 'only']:
print('Downloading comments...')
danmaku_output=self.save_dir/f'{filename}.json'
return_value['danmaku'] =danmaku_output
ifcomments=='yes':
ex.submit(self.download_comments_native, message_server_info, danmaku_output)
elifcomments=='only':
self.download_comments_native(message_server_info, danmaku_output)
returnreturn_value
else:
return_value['danmaku'] =None
master_m3u8_url=stream_info['data']['uri']
if'assetdelivery.dlive'inmaster_m3u8_url:
print('WARN: This is a DLive stream. Will use yt-dlp to download.')
assert'cookies'instream_info['data']
forcinstream_info['data']['cookies']:
self.session.cookies.set(c['name'], c['value'])
playlist_url=None
ifverbose:
print('master m3u8 URL:', master_m3u8_url)
print('================== content ==================')
print(self.session.get(master_m3u8_url).text)
print('==================== end ====================')
output=self.save_dir/f'{filename}.mp4'
dlive_bid=self.session.cookies.get("dlive_bid")
# cmd = f'yt-dlp "{master_m3u8_url}" --ignore-config -N 10 -o "{output}" --add-headers "Cookie:dlive_bid={dlive_bid}"'
save_dir_str=str(self.save_dir).rstrip('\\') # remove trailing backslash otherwise it will escape quotes in cmd
cmd=f'N_m3u8DL-RE "{master_m3u8_url}" --save-name "{filename}" --save-dir "{save_dir_str}" --auto-select -H "Cookie:dlive_bid={dlive_bid}" -mt -M format=mp4 --no-date-info'
else:
master_m3u8_text=self.session.get(master_m3u8_url).text
playlist_url=re.search(r'^.+playlist\.m3u8.*$', master_m3u8_text, re.MULTILINE)[0]
playlist_url=urljoin(master_m3u8_url, playlist_url) #+ '&start=682.251'
ifverbose:
print('master m3u8 URL:', master_m3u8_url)
print('================== content ==================')
print(master_m3u8_text)
print('==================== end ====================')
print('playlist m3u8 URL:', playlist_url)
print('================== content ==================')
print(self.session.get(playlist_url).text)
print('==================== end ====================')
output=self.save_dir/f'{filename}.ts'
# do not use arrays. the way python quotes & is not compatible with cmd/bat which minyami uses.
# See: https://stackoverflow.com/questions/74700723/
# Make sure to also use shell=True for *nix systems
cmd=f'minyami -d "{playlist_url}" --key {audience_token},{max_quality} -o "{output}"'
ifself.proxy:
cmd+=f' --proxy "{self.proxy}"'
ifverbose:
cmd+=' --verbose'
print('CMD is:')
print(cmd)
ifsimulate:
output=None
else:
run(cmd, shell=True)
ex.shutdown(wait=True) # ensure download_comments is finished
return_value.update({
'master_m3u8_url': master_m3u8_url,
'playlist_m3u8_url': playlist_url,
'output': output,
})
returnreturn_value
defdownload_thumbnail(self, url_or_video_id, info_only=False, dump=False):
video_id, url, video_type=self._parse_url_or_video_id(url_or_video_id)
ifvideo_type=='live':
print('Cannot download thumbnail for live.')
return
soup=get(url, session=self.session)
data=json.loads(soup.find(id="js-initial-watch-data")["data-api-data"])
ifdump:
dump_json(data, self.save_dir/f'{video_id}.info.json')
ifinfo_only:
return
thumbnails=data["video"]["thumbnail"]
# get the last value, which is the highest resolution
name, thumbnail_url=list(thumbnails.items())[-1]
print(f"Best thumbnail variant: {name}, {thumbnail_url}")
download(thumbnail_url, filename=self.save_dir/video_id)
if__name__=="__main__":
importshlex
importsys
importargparse
# auto load arguments from nico.txt
forfin [Path(__file__).parent/'nico.txt', Path('nico.txt')]:
iff.exists():
withopen(f, encoding='utf8') asf:
# insert in front so it can be overridden
commands=shlex.split(f.read().replace('\\', '\\\\'))
sys.argv[1:1] =commands
break
# https://stackoverflow.com/questions/3853722/how-to-insert-newlines-on-argparse-help-text
classSmartFormatter(argparse.HelpFormatter):
def_split_lines(self, text, width):
iftext.startswith('R|'):
returntext[2:].splitlines()
returnargparse.HelpFormatter._split_lines(self, text, width)
parser=argparse.ArgumentParser(formatter_class=SmartFormatter)
parser.add_argument("url", help="URL or ID of nicovideo webpage")
parser.add_argument('--verbose', '-v', action='store_true', help='Print verbose info for debugging.')
parser.add_argument('--info', '-i', action='store_true', help='Print info only.')
parser.add_argument('--dump', action='store_true', help='Dump all the metadata to json files.')
parser.add_argument('--thumb', action='store_true', help='Download thumbnail only. Only works for video type (not live type).')
parser.add_argument('--cookies', '-c', help='R|Cookie source.\nProvide either:\n - A browser name to fetch from;\n - The value of "user_session";\n - A Netscape-style cookie file.')
parser.add_argument('--comments', '-d', default='no', choices=['yes', 'no', 'only'], help='Control if comments (danmaku) are downloaded. [Default: no]')
parser.add_argument('--proxy', default='auto', help='Specify a proxy, "none", or "auto" (automatically detects system proxy settings). [Default: auto]')
parser.add_argument('--save-dir', '-o', help='Specify the directory to save the downloaded files. [Default: current directory]')
parser.add_argument('--reserve', action='store_true', help='Automatically reserve timeshift ticket if not reserved yet. [Default: no]')
parser.add_argument('--simulate', action='store_true', help='Simulate the download process without actually downloading.')
args=parser.parse_args()
nico_downloader=NicoDownloader(args.cookies, args.proxy, save_dir=args.save_dir)
ifargs.thumb:
nico_downloader.download_thumbnail(args.url, info_only=args.info, dump=args.dump)
else:
nico_downloader.download_timeshift(args.url, info_only=args.info, verbose=args.verbose, comments=args.comments, dump=args.dump, auto_reserve=args.reserve, simulate=args.simulate)