MCPcopy Create free account
hub / github.com/nathom/streamrip / file

Function file

streamrip/rip/cli.py:221–264  ·  view source on GitHub ↗

Download content from URLs in a file. Example usage: rip file urls.txt

(ctx, path)

Source from the content-addressed store, hash-verified

219@click.pass_context
220@coro
221async def file(ctx, path):
222 """Download content from URLs in a file.
223
224 Example usage:
225
226 rip file urls.txt
227 """
228 try:
229 with ctx.obj["config"] as cfg:
230 async with Main(cfg) as main:
231 async with aiofiles.open(path, "r") as f:
232 content = await f.read()
233 try:
234 items: Any = json.loads(content)
235 loaded = True
236 except json.JSONDecodeError:
237 items = content.split()
238 loaded = False
239 if loaded:
240 console.print(
241 f"Detected json file. Loading [yellow]{len(items)}[/yellow] items"
242 )
243 await main.add_all_by_id(
244 [(i["source"], i["media_type"], i["id"]) for i in items]
245 )
246 else:
247 s = set(items)
248 if len(s) < len(items):
249 console.print(
250 f"Found [orange]{len(items)-len(s)}[/orange] repeated URLs!"
251 )
252 items = list(s)
253 console.print(
254 f"Detected list of urls. Loading [yellow]{len(items)}[/yellow] items"
255 )
256 await main.add_all(items)
257
258 await main.resolve()
259 await main.rip()
260 except aiohttp.ClientConnectorCertificateError as e:
261 from ..utils.ssl_utils import print_ssl_error_help
262
263 console.print(f"[red]SSL Certificate verification error: {e}[/red]")
264 print_ssl_error_help()
265
266
267@rip.group()

Callers

nothing calls this directly

Calls 7

MainClass · 0.85
print_ssl_error_helpFunction · 0.85
printMethod · 0.80
add_all_by_idMethod · 0.80
add_allMethod · 0.80
resolveMethod · 0.45
ripMethod · 0.45

Tested by

no test coverage detected